Merge pull request #847 from krakend/dev-2.6

Bump version to 2.6.0

Dockerfile

@@ -14,7 +14,7 @@ FROM alpine:${ALPINE_VERSION}
 
 LABEL maintainer="[email protected]"
 
-RUN apk add --no-cache ca-certificates tzdata && \
+RUN apk upgrade --no-cache --no-interactive && apk add --no-cache ca-certificates tzdata && \
     adduser -u 1000 -S -D -H krakend && \
     mkdir /etc/krakend && \
     echo '{ "version": 3 }' > /etc/krakend/krakend.json

Makefile

@@ -6,7 +6,7 @@
 
 BIN_NAME :=krakend
 OS := $(shell uname | tr '[:upper:]' '[:lower:]')
-VERSION := 2.5.1
+VERSION := 2.6.0
 SCHEMA_VERSION := $(shell echo "${VERSION}" | cut -d '.' -f 1,2)
 GIT_COMMIT := $(shell git rev-parse --short=7 HEAD)
 PKGNAME := krakend
@@ -20,7 +20,7 @@ DESC := High performance API gateway. Aggregate, filter, manipulate and add midd
 MAINTAINER := Daniel Ortiz <[email protected]>
 DOCKER_WDIR := /tmp/fpm
 DOCKER_FPM := devopsfaith/fpm
-GOLANG_VERSION := 1.20.13
+GOLANG_VERSION := 1.20.14
 GLIBC_VERSION := $(shell sh find_glibc.sh)
 ALPINE_VERSION := 3.18
 OS_TAG :=

backend_factory.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	otellura "github.com/krakend/krakend-otel/lura"
 	amqp "github.com/krakendio/krakend-amqp/v2"
 	cel "github.com/krakendio/krakend-cel/v2"
 	cb "github.com/krakendio/krakend-circuitbreaker/v2/gobreaker/proxy"
@@ -39,18 +40,25 @@ func NewBackendFactory(logger logging.Logger, metricCollector *metrics.Metrics)
 	return NewBackendFactoryWithContext(context.Background(), logger, metricCollector)
 }
 
-// NewBackendFactoryWithContext creates a BackendFactory by stacking all the available middlewares and injecting the received context
-func NewBackendFactoryWithContext(ctx context.Context, logger logging.Logger, metricCollector *metrics.Metrics) proxy.BackendFactory {
+func newRequestExecutorFactory(logger logging.Logger) func(*config.Backend) client.HTTPRequestExecutor {
 	requestExecutorFactory := func(cfg *config.Backend) client.HTTPRequestExecutor {
 		clientFactory := client.NewHTTPClient
 		if _, ok := cfg.ExtraConfig[oauth2client.Namespace]; ok {
 			clientFactory = oauth2client.NewHTTPClient(cfg)
 		} else {
 			clientFactory = httpcache.NewHTTPClient(cfg, clientFactory)
 		}
+
+		clientFactory = otellura.InstrumentedHTTPClientFactory(clientFactory, cfg)
+		// TODO: check what happens if we have both, opencensus and otel enabled ?
 		return opencensus.HTTPRequestExecutorFromConfig(clientFactory, cfg)
 	}
-	requestExecutorFactory = httprequestexecutor.HTTPRequestExecutor(logger, requestExecutorFactory)
+	return httprequestexecutor.HTTPRequestExecutor(logger, requestExecutorFactory)
+}
+
+func internalNewBackendFactory(ctx context.Context, requestExecutorFactory func(*config.Backend) client.HTTPRequestExecutor,
+	logger logging.Logger, metricCollector *metrics.Metrics) proxy.BackendFactory {
+
 	backendFactory := martian.NewConfiguredBackendFactory(logger, requestExecutorFactory)
 	bf := pubsub.NewBackendFactory(ctx, logger, backendFactory)
 	backendFactory = bf.New
@@ -62,13 +70,19 @@ func NewBackendFactoryWithContext(ctx context.Context, logger logging.Logger, me
 	backendFactory = cb.BackendFactory(backendFactory, logger)
 	backendFactory = metricCollector.BackendFactory("backend", backendFactory)
 	backendFactory = opencensus.BackendFactory(backendFactory)
-
+	backendFactory = otellura.BackendFactory(backendFactory)
 	return func(remote *config.Backend) proxy.Proxy {
 		logger.Debug(fmt.Sprintf("[BACKEND: %s] Building the backend pipe", remote.URLPattern))
 		return backendFactory(remote)
 	}
 }
 
+// NewBackendFactoryWithContext creates a BackendFactory by stacking all the available middlewares and injecting the received context
+func NewBackendFactoryWithContext(ctx context.Context, logger logging.Logger, metricCollector *metrics.Metrics) proxy.BackendFactory {
+	requestExecutorFactory := newRequestExecutorFactory(logger)
+	return internalNewBackendFactory(ctx, requestExecutorFactory, logger, metricCollector)
+}
+
 type backendFactory struct{}
 
 func (backendFactory) NewBackendFactory(ctx context.Context, l logging.Logger, m *metrics.Metrics) proxy.BackendFactory {

cmd/krakend-ce/main.go

@@ -57,6 +57,18 @@ func main() {
 		})
 	}
 
+	commandsToLoad := []cmd.Command{
+		cmd.RunCommand,
+		cmd.CheckCommand,
+		cmd.PluginCommand,
+		cmd.VersionCommand,
+		cmd.AuditCommand,
+		krakend.NewTestPluginCmd(),
+	}
+
+	cmd.DefaultRoot = cmd.NewRoot(cmd.RootCommand, commandsToLoad...)
+	cmd.DefaultRoot.Cmd.CompletionOptions.DisableDefaultCmd = true
+
 	cmd.Execute(cfg, krakend.NewExecutor(ctx))
 }
 

executor.go

@@ -12,6 +12,9 @@ import (
 	"github.com/go-contrib/uuid"
 	"golang.org/x/sync/errgroup"
 
+	kotel "github.com/krakend/krakend-otel"
+	otellura "github.com/krakend/krakend-otel/lura"
+	otelgin "github.com/krakend/krakend-otel/router/gin"
 	krakendbf "github.com/krakendio/bloomfilter/v2/krakend"
 	asyncamqp "github.com/krakendio/krakend-amqp/v2/async"
 	audit "github.com/krakendio/krakend-audit"
@@ -53,10 +56,16 @@ func NewExecutor(ctx context.Context) cmd.Executor {
 }
 
 // PluginLoader defines the interface for the collaborator responsible of starting the plugin loaders
+// Deprecated: Use PluginLoaderWithContext
 type PluginLoader interface {
 	Load(folder, pattern string, logger logging.Logger)
 }
 
+// PluginLoaderWithContext defines the interface for the collaborator responsible of starting the plugin loaders
+type PluginLoaderWithContext interface {
+	LoadWithContext(ctx context.Context, folder, pattern string, logger logging.Logger)
+}
+
 // SubscriberFactoriesRegister registers all the required subscriber factories from the available service
 // discover components and adapters and returns a service register function.
 // The service register function will register the service by the given name and port to all the available
@@ -123,17 +132,20 @@ type AgentStarter interface {
 
 // ExecutorBuilder is a composable builder. Every injected property is used by the NewCmdExecutor method.
 type ExecutorBuilder struct {
-	LoggerFactory               LoggerFactory
+	// PluginLoader is deprecated: Use PluginLoaderWithContext
 	PluginLoader                PluginLoader
+	PluginLoaderWithContext     PluginLoaderWithContext
+	LoggerFactory               LoggerFactory
 	SubscriberFactoriesRegister SubscriberFactoriesRegister
 	TokenRejecterFactory        TokenRejecterFactory
 	MetricsAndTracesRegister    MetricsAndTracesRegister
 	EngineFactory               EngineFactory
-	ProxyFactory                ProxyFactory
-	BackendFactory              BackendFactory
-	HandlerFactory              HandlerFactory
-	RunServerFactory            RunServerFactory
-	AgentStarterFactory         AgentStarter
+
+	ProxyFactory        ProxyFactory
+	BackendFactory      BackendFactory
+	HandlerFactory      HandlerFactory
+	RunServerFactory    RunServerFactory
+	AgentStarterFactory AgentStarter
 
 	Middlewares []gin.HandlerFunc
 }
@@ -161,10 +173,13 @@ func (e *ExecutorBuilder) NewCmdExecutor(ctx context.Context) cmd.Executor {
 		}
 
 		if cfg.Plugin != nil {
-			e.PluginLoader.Load(cfg.Plugin.Folder, cfg.Plugin.Pattern, logger)
+			e.PluginLoaderWithContext.LoadWithContext(ctx, cfg.Plugin.Folder, cfg.Plugin.Pattern, logger)
 		}
 
 		metricCollector := e.MetricsAndTracesRegister.Register(ctx, cfg, logger)
+		if metricsAndTracesCloser, ok := e.MetricsAndTracesRegister.(io.Closer); ok {
+			defer metricsAndTracesCloser.Close()
+		}
 
 		// Initializes the global cache for the JWK clients if enabled in the config
 		if err := jose.SetGlobalCacher(logger, cfg.ExtraConfig); err != nil && err != jose.ErrNoValidatorCfg {
@@ -180,14 +195,18 @@ func (e *ExecutorBuilder) NewCmdExecutor(ctx context.Context) cmd.Executor {
 			logger.Warning("[SERVICE: Bloomfilter]", err.Error())
 		}
 
-		pf := e.ProxyFactory.NewProxyFactory(
-			logger,
-			e.BackendFactory.NewBackendFactory(ctx, logger, metricCollector),
-			metricCollector,
-		)
+		bpf := e.BackendFactory.NewBackendFactory(ctx, logger, metricCollector)
+		pf := e.ProxyFactory.NewProxyFactory(logger, bpf, metricCollector)
 
 		agentPing := make(chan string, len(cfg.AsyncAgents))
 
+		handlerF := e.HandlerFactory.NewHandlerFactory(logger, metricCollector, tokenRejecterFactory)
+		handlerF = otelgin.New(handlerF)
+
+		runServerChain := serverhttp.RunServerWithLoggerFactory(logger)
+		runServerChain = otellura.GlobalRunServer(logger, runServerChain)
+		runServerChain = router.RunServerFunc(e.RunServerFactory.NewRunServer(logger, runServerChain))
+
 		// setup the krakend router
 		routerFactory := router.NewFactory(router.Config{
 			Engine: e.EngineFactory.NewEngine(cfg, router.EngineOptions{
@@ -198,8 +217,8 @@ func (e *ExecutorBuilder) NewCmdExecutor(ctx context.Context) cmd.Executor {
 			ProxyFactory:   pf,
 			Middlewares:    e.Middlewares,
 			Logger:         logger,
-			HandlerFactory: e.HandlerFactory.NewHandlerFactory(logger, metricCollector, tokenRejecterFactory),
-			RunServer:      router.RunServerFunc(e.RunServerFactory.NewRunServer(logger, serverhttp.RunServerWithLoggerFactory(logger))),
+			HandlerFactory: handlerF,
+			RunServer:      runServerChain,
 		})
 
 		// start the engines
@@ -238,6 +257,9 @@ func (e *ExecutorBuilder) checkCollaborators() {
 	if e.PluginLoader == nil {
 		e.PluginLoader = new(pluginLoader)
 	}
+	if e.PluginLoaderWithContext == nil {
+		e.PluginLoaderWithContext = new(pluginLoader)
+	}
 	if e.SubscriberFactoriesRegister == nil {
 		e.SubscriberFactoriesRegister = new(registerSubscriberFactories)
 	}
@@ -352,10 +374,12 @@ func (BloomFilterJWT) NewTokenRejecter(ctx context.Context, cfg config.ServiceCo
 }
 
 // MetricsAndTraces is the default implementation of the MetricsAndTracesRegister interface.
-type MetricsAndTraces struct{}
+type MetricsAndTraces struct {
+	shutdownFn func()
+}
 
 // Register registers the metrics, influx and opencensus packages as required by the given configuration.
-func (MetricsAndTraces) Register(ctx context.Context, cfg config.ServiceConfig, l logging.Logger) *metrics.Metrics {
+func (m *MetricsAndTraces) Register(ctx context.Context, cfg config.ServiceConfig, l logging.Logger) *metrics.Metrics {
 	metricCollector := metrics.New(ctx, cfg.ExtraConfig, l)
 
 	if err := influxdb.New(ctx, cfg.ExtraConfig, metricCollector, l); err != nil {
@@ -374,9 +398,21 @@ func (MetricsAndTraces) Register(ctx context.Context, cfg config.ServiceConfig,
 		l.Debug("[SERVICE: OpenCensus] Service correctly registered")
 	}
 
+	if shutdownFn, err := kotel.Register(ctx, l, cfg); err == nil {
+		m.shutdownFn = shutdownFn
+	} else {
+		l.Error(fmt.Sprintf("[SERVICE: OpenTelemetry] cannot register exporters: %s", err.Error()))
+	}
+
 	return metricCollector
 }
 
+func (m *MetricsAndTraces) Close() {
+	if m.shutdownFn != nil {
+		m.shutdownFn()
+	}
+}
+
 const (
 	usageDisable = "USAGE_DISABLE"
 	usageDelay   = 5 * time.Second