perf: Tracing observability tool integration

App.go

@@ -22,6 +22,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"github.com/devtron-labs/devtron/client/telemetry"
+	"github.com/devtron-labs/devtron/otel"
 	"log"
 	"net/http"
 	"os"
@@ -36,6 +37,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/user"
 	"github.com/go-pg/pg"
 	_ "github.com/lib/pq"
+	"go.opentelemetry.io/contrib/instrumentation/github.com/gorilla/mux/otelmux"
 	"go.uber.org/zap"
 )
 
@@ -49,8 +51,9 @@ type App struct {
 	pubsubClient  *pubsub.PubSubClient
 	posthogClient *telemetry.PosthogClient
 	// used for local dev only
-	serveTls        bool
-	sessionManager2 *authMiddleware.SessionManager
+	serveTls           bool
+	sessionManager2    *authMiddleware.SessionManager
+	OtelTracingService *otel.OtelTracingServiceImpl
 }
 
 func NewApp(router *router.MuxRouter,
@@ -65,15 +68,16 @@ func NewApp(router *router.MuxRouter,
 	//check argo connection
 	//todo - check argo-cd version on acd integration installation
 	app := &App{
-		MuxRouter:       router,
-		Logger:          Logger,
-		SSE:             sse,
-		Enforcer:        enforcer,
-		db:              db,
-		pubsubClient:    pubsubClient,
-		serveTls:        false,
-		sessionManager2: sessionManager2,
-		posthogClient:   posthogClient,
+		MuxRouter:          router,
+		Logger:             Logger,
+		SSE:                sse,
+		Enforcer:           enforcer,
+		db:                 db,
+		pubsubClient:       pubsubClient,
+		serveTls:           false,
+		sessionManager2:    sessionManager2,
+		posthogClient:      posthogClient,
+		OtelTracingService: otel.NewOtelTracingServiceImpl(Logger),
 	}
 	return app
 }
@@ -82,11 +86,19 @@ func (app *App) Start() {
 	port := 8080 //TODO: extract from environment variable
 	app.Logger.Debugw("starting server")
 	app.Logger.Infow("starting server on ", "port", port)
+
+	// setup tracer
+	tracerProvider := app.OtelTracingService.Init(otel.OTEL_ORCHESTRASTOR_SERVICE_NAME)
+
 	app.MuxRouter.Init()
 	//authEnforcer := casbin2.Create()
 
 	server := &http.Server{Addr: fmt.Sprintf(":%d", port), Handler: authMiddleware.Authorizer(app.sessionManager2, user.WhitelistChecker)(app.MuxRouter.Router)}
+
 	app.MuxRouter.Router.Use(middleware.PrometheusMiddleware)
+	if tracerProvider != nil {
+		app.MuxRouter.Router.Use(otelmux.Middleware(otel.OTEL_ORCHESTRASTOR_SERVICE_NAME))
+	}
 	app.server = server
 	var err error
 	if app.serveTls {
@@ -124,6 +136,9 @@ func (app *App) Stop() {
 	if err != nil {
 		app.Logger.Errorw("error in mux router shutdown", "err", err)
 	}
+
+	app.OtelTracingService.Shutdown()
+
 	app.Logger.Infow("closing db connection")
 	err = app.db.Close()
 	if err != nil {

Wire.go

@@ -336,6 +336,8 @@ func InitializeApp() (*App, error) {
 		//ArgoUtil.NewClusterServiceImpl,
 		//wire.Bind(new(ArgoUtil.ClusterService), new(ArgoUtil.ClusterServiceImpl)),
 		pipeline.GetEcrConfig,
+		//otel.NewOtelTracingServiceImpl,
+		//wire.Bind(new(otel.OtelTracingService), new(*otel.OtelTracingServiceImpl)),
 		NewApp,
 		//session.NewK8sClient,
 

api/helm-app/HelmAppRestHandler.go

@@ -87,7 +87,7 @@ func (handler *HelmAppRestHandlerImpl) ListApplications(w http.ResponseWriter, r
 		clusterIds = append(clusterIds, j)
 	}
 	token := r.Header.Get("token")
-	handler.helmAppService.ListHelmApplications(clusterIds, w, token, handler.checkHelmAuth)
+	handler.helmAppService.ListHelmApplications(r.Context(), clusterIds, w, token, handler.checkHelmAuth)
 }
 
 func (handler *HelmAppRestHandlerImpl) GetApplicationDetail(w http.ResponseWriter, r *http.Request) {
@@ -153,7 +153,7 @@ func (handler *HelmAppRestHandlerImpl) Hibernate(w http.ResponseWriter, r *http.
 		return
 	}
 	//RBAC enforcer Ends
-	res, err := handler.helmAppService.HibernateApplication(context.Background(), appIdentifier, hibernateRequest)
+	res, err := handler.helmAppService.HibernateApplication(r.Context(), appIdentifier, hibernateRequest)
 	if err != nil {
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
 		return
@@ -185,7 +185,7 @@ func (handler *HelmAppRestHandlerImpl) UnHibernate(w http.ResponseWriter, r *htt
 		return
 	}
 	//RBAC enforcer Ends
-	res, err := handler.helmAppService.UnHibernateApplication(context.Background(), appIdentifier, hibernateRequest)
+	res, err := handler.helmAppService.UnHibernateApplication(r.Context(), appIdentifier, hibernateRequest)
 	if err != nil {
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
 		return
@@ -212,7 +212,7 @@ func (handler *HelmAppRestHandlerImpl) GetReleaseInfo(w http.ResponseWriter, r *
 		return
 	}
 	//RBAC enforcer Ends
-	releaseInfo, err := handler.helmAppService.GetValuesYaml(context.Background(), appIdentifier)
+	releaseInfo, err := handler.helmAppService.GetValuesYaml(r.Context(), appIdentifier)
 	if err != nil {
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
 		return
@@ -255,7 +255,7 @@ func (handler *HelmAppRestHandlerImpl) GetDesiredManifest(w http.ResponseWriter,
 		return
 	}
 	//RBAC enforcer Ends
-	res, err := handler.helmAppService.GetDesiredManifest(context.Background(), appIdentifier, desiredManifestRequest.Resource)
+	res, err := handler.helmAppService.GetDesiredManifest(r.Context(), appIdentifier, desiredManifestRequest.Resource)
 	if err != nil {
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
 		return
@@ -304,7 +304,7 @@ func (handler *HelmAppRestHandlerImpl) DeleteApplication(w http.ResponseWriter,
 		return
 	}
 
-	res, err := handler.helmAppService.DeleteApplication(context.Background(), appIdentifier)
+	res, err := handler.helmAppService.DeleteApplication(r.Context(), appIdentifier)
 	if err != nil {
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
 		return
@@ -338,7 +338,7 @@ func (handler *HelmAppRestHandlerImpl) UpdateApplication(w http.ResponseWriter,
 	//RBAC enforcer Ends
 
 	// update application externally
-	res, err := handler.helmAppService.UpdateApplication(context.Background(), appIdentifier, request)
+	res, err := handler.helmAppService.UpdateApplication(r.Context(), appIdentifier, request)
 	if err != nil {
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
 		return
@@ -364,7 +364,7 @@ func (handler *HelmAppRestHandlerImpl) TemplateChart(w http.ResponseWriter, r *h
 	//making this api rbac free
 
 	// template chart starts
-	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
 	defer cancel()
 	response, err := handler.helmAppService.TemplateChart(ctx, request)
 	if err != nil {

api/helm-app/HelmAppService.go

@@ -24,6 +24,7 @@ import (
 	"github.com/gogo/protobuf/proto"
 	"github.com/tidwall/gjson"
 	"github.com/tidwall/sjson"
+	"go.opentelemetry.io/otel"
 	"go.uber.org/zap"
 	"net/http"
 	"reflect"
@@ -36,7 +37,7 @@ const DEFAULT_CLUSTER = "default_cluster"
 const DEFAULT_CLUSTER_ID = 1
 
 type HelmAppService interface {
-	ListHelmApplications(clusterIds []int, w http.ResponseWriter, token string, helmAuth func(token string, object string) bool)
+	ListHelmApplications(ctx context.Context, clusterIds []int, w http.ResponseWriter, token string, helmAuth func(token string, object string) bool)
 	GetApplicationDetail(ctx context.Context, app *AppIdentifier) (*AppDetail, error)
 	GetApplicationDetailWithFilter(ctx context.Context, app *AppIdentifier, resourceTreeFilter *ResourceTreeFilter) (*AppDetail, error)
 	HibernateApplication(ctx context.Context, app *AppIdentifier, hibernateRequest *openapi.HibernateRequest) ([]*openapi.HibernateStatus, error)
@@ -101,11 +102,13 @@ type ResourceRequestBean struct {
 	K8sRequest application.K8sRequestBean `json:"k8sRequest"`
 }
 
-func (impl *HelmAppServiceImpl) listApplications(clusterIds []int) (ApplicationService_ListApplicationsClient, error) {
+func (impl *HelmAppServiceImpl) listApplications(ctx context.Context, clusterIds []int) (ApplicationService_ListApplicationsClient, error) {
 	if len(clusterIds) == 0 {
 		return nil, nil
 	}
+	_, span := otel.Tracer("clusterService").Start(ctx, "FindByIds")
 	clusters, err := impl.clusterService.FindByIds(clusterIds)
+	span.End()
 	if err != nil {
 		impl.logger.Errorw("error in fetching cluster detail", "err", err)
 		return nil, err
@@ -120,31 +123,35 @@ func (impl *HelmAppServiceImpl) listApplications(clusterIds []int) (ApplicationS
 		}
 		req.Clusters = append(req.Clusters, config)
 	}
-	applicatonStream, err := impl.helmAppClient.ListApplication(req)
+	applicatonStream, err := impl.helmAppClient.ListApplication(ctx, req)
 	if err != nil {
 		return nil, err
 	}
 
 	return applicatonStream, err
 }
 
-func (impl *HelmAppServiceImpl) ListHelmApplications(clusterIds []int, w http.ResponseWriter, token string, helmAuth func(token string, object string) bool) {
+func (impl *HelmAppServiceImpl) ListHelmApplications(ctx context.Context, clusterIds []int, w http.ResponseWriter, token string, helmAuth func(token string, object string) bool) {
 	var helmCdPipelines []*pipelineConfig.Pipeline
 	var installedHelmApps []*repository.InstalledApps
-	appStream, err := impl.listApplications(clusterIds)
+	appStream, err := impl.listApplications(ctx, clusterIds)
 	if err != nil {
 		impl.logger.Errorw("error in fetching app list", "clusters", clusterIds, "err", err)
 	}
 	if err == nil && len(clusterIds) > 0 {
 		// get helm apps which are created using cd_pipelines
+		newCtx, span := otel.Tracer("pipelineRepository").Start(ctx, "GetAppAndEnvDetailsForDeploymentAppTypePipeline")
 		helmCdPipelines, err = impl.pipelineRepository.GetAppAndEnvDetailsForDeploymentAppTypePipeline(util.PIPELINE_DEPLOYMENT_TYPE_HELM, clusterIds)
+		span.End()
 		if err != nil {
 			impl.logger.Errorw("error in fetching helm app list from DB created using cd_pipelines", "clusters", clusterIds, "err", err)
 		}
 
 		// if not hyperion mode, then fetch from installed_apps whose deployment_app_type is helm (as in hyperion mode, these apps should be treated as external-apps)
 		if !util2.IsBaseStack() {
+			newCtx, span = otel.Tracer("pipelineRepository").Start(newCtx, "GetAppAndEnvDetailsForDeploymentAppTypePipeline")
 			installedHelmApps, err = impl.installedAppRepository.GetAppAndEnvDetailsForDeploymentAppTypeInstalledApps(util.PIPELINE_DEPLOYMENT_TYPE_HELM, clusterIds)
+			span.End()
 			if err != nil {
 				impl.logger.Errorw("error in fetching helm app list from DB created from app store", "clusters", clusterIds, "err", err)
 			}

api/helm-app/applicationClient.go

@@ -4,13 +4,14 @@ import (
 	"context"
 	"fmt"
 	"github.com/caarlos0/env"
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"time"
 )
 
 type HelmAppClient interface {
-	ListApplication(req *AppListRequest) (ApplicationService_ListApplicationsClient, error)
+	ListApplication(ctx context.Context, req *AppListRequest) (ApplicationService_ListApplicationsClient, error)
 	GetAppDetail(ctx context.Context, in *AppDetailRequest) (*AppDetail, error)
 	Hibernate(ctx context.Context, in *HibernateRequest) (*HibernateResponse, error)
 	UnHibernate(ctx context.Context, in *HibernateRequest) (*HibernateResponse, error)
@@ -66,6 +67,8 @@ func (impl *HelmAppClientImpl) getConnection() (*grpc.ClientConn, error) {
 	var opts []grpc.DialOption
 	ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
 	opts = append(opts,
+		grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()),
+		grpc.WithStreamInterceptor(otelgrpc.StreamClientInterceptor()),
 		grpc.WithBlock(),
 		grpc.WithInsecure(),
 		grpc.WithDefaultCallOptions(
@@ -81,12 +84,12 @@ func (impl *HelmAppClientImpl) getConnection() (*grpc.ClientConn, error) {
 	return conn, err
 }
 
-func (impl *HelmAppClientImpl) ListApplication(req *AppListRequest) (ApplicationService_ListApplicationsClient, error) {
+func (impl *HelmAppClientImpl) ListApplication(ctx context.Context, req *AppListRequest) (ApplicationService_ListApplicationsClient, error) {
 	applicationClient, err := impl.getApplicationClient()
 	if err != nil {
 		return nil, err
 	}
-	stream, err := applicationClient.ListApplications(context.Background(), req)
+	stream, err := applicationClient.ListApplications(ctx, req)
 	if err != nil {
 		return nil, err
 	}