feat: add OTEL instrumentation for authentication and handlers

go.mod

@@ -120,6 +120,11 @@ require (
 	sigs.k8s.io/yaml v1.6.0
 )
 
+require (
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0
+	go.opentelemetry.io/otel/trace v1.39.0
+)
+
 require (
 	cloud.google.com/go/auth v0.15.0 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.7 // indirect
@@ -273,10 +278,8 @@ require (
 	github.com/xlab/treeprint v1.2.0 // indirect
 	go.mongodb.org/mongo-driver v1.17.6 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
-	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 // indirect
 	go.opentelemetry.io/otel/metric v1.39.0 // indirect
-	go.opentelemetry.io/otel/trace v1.39.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.9.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect

server/server.go

@@ -44,8 +44,12 @@ import (
 	log "github.com/sirupsen/logrus"
 	"github.com/soheilhy/cmux"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	otel_codes "go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
@@ -166,13 +170,17 @@ var (
 	enableGRPCTimeHistogram         = true
 )
 
+// OpenTelemetry tracer for this package
+var tracer trace.Tracer
+
 func init() {
 	maxConcurrentLoginRequestsCount = env.ParseNumFromEnv(maxConcurrentLoginRequestsCountEnv, maxConcurrentLoginRequestsCount, 0, math.MaxInt32)
 	replicasCount = env.ParseNumFromEnv(replicasCountEnv, replicasCount, 0, math.MaxInt32)
 	if replicasCount > 0 {
 		maxConcurrentLoginRequestsCount = maxConcurrentLoginRequestsCount / replicasCount
 	}
 	enableGRPCTimeHistogram = env.ParseBoolFromEnv(common.EnvEnableGRPCTimeHistogramEnv, false)
+	tracer = otel.Tracer("github.com/argoproj/argo-cd/v3/server")
 }
 
 // ArgoCDServer is the API server for Argo CD
@@ -1176,8 +1184,8 @@ func (server *ArgoCDServer) newHTTPServer(ctx context.Context, port int, grpcWeb
 		Handler: &handlerSwitcher{
 			handler: mux,
 			urlToHandler: map[string]http.Handler{
-				"/api/badge":          badge.NewHandler(server.AppClientset, server.settingsMgr, server.Namespace, server.ApplicationNamespaces),
-				common.LogoutEndpoint: logout.NewHandler(server.settingsMgr, server.sessionMgr, server.RootPath, server.BaseHRef),
+				"/api/badge":          otelhttp.NewHandler(badge.NewHandler(server.AppClientset, server.settingsMgr, server.Namespace, server.ApplicationNamespaces), "server.ArgoCDServer/badge"),
+				common.LogoutEndpoint: otelhttp.NewHandler(logout.NewHandler(server.settingsMgr, server.sessionMgr, server.RootPath, server.BaseHRef), "server.ArgoCDServer/logout"),
 			},
 			contentTypeToHandler: map[string]http.Handler{
 				"application/grpc-web+proto": grpcWebHandler,
@@ -1305,7 +1313,7 @@ func registerExtensions(mux *http.ServeMux, a *ArgoCDServer, metricsReg HTTPMetr
 	extHandler := http.HandlerFunc(a.extensionManager.CallExtension())
 	authMiddleware := a.sessionMgr.AuthMiddlewareFunc(a.DisableAuth, a.settings.IsSSOConfigured(), a.ssoClientApp)
 	// auth middleware ensures that requests to all extensions are authenticated first
-	mux.Handle(extension.URLPrefix+"/", authMiddleware(extHandler))
+	mux.Handle(extension.URLPrefix+"/", otelhttp.NewHandler(authMiddleware(extHandler), "server.ArgoCDServer/extensions"))
 
 	a.extensionManager.AddMetricsRegistry(metricsReg)
 
@@ -1363,9 +1371,10 @@ func (server *ArgoCDServer) registerDexHandlers(mux *http.ServeMux) {
 		return
 	}
 	// Run dex OpenID Connect Identity Provider behind a reverse proxy (served at /api/dex)
-	mux.HandleFunc(common.DexAPIEndpoint+"/", dexutil.NewDexHTTPReverseProxy(server.DexServerAddr, server.BaseHRef, server.DexTLSConfig))
-	mux.HandleFunc(common.LoginEndpoint, server.ssoClientApp.HandleLogin)
-	mux.HandleFunc(common.CallbackEndpoint, server.ssoClientApp.HandleCallback)
+	mux.Handle(common.DexAPIEndpoint+"/", otelhttp.NewHandler(http.HandlerFunc(dexutil.NewDexHTTPReverseProxy(server.DexServerAddr, server.BaseHRef, server.DexTLSConfig)), "server.dex/Proxy"))
+
+	mux.Handle(common.LoginEndpoint, otelhttp.NewHandler(http.HandlerFunc(server.ssoClientApp.HandleLogin), "server.ClientApp/HandleLogin"))
+	mux.Handle(common.CallbackEndpoint, otelhttp.NewHandler(http.HandlerFunc(server.ssoClientApp.HandleCallback), "server.ClientApp/HandleCallback"))
 }
 
 // newRedirectServer returns an HTTP server which does a 307 redirect to the HTTPS server
@@ -1522,6 +1531,9 @@ func replaceBaseHRef(data string, replaceWith string) string {
 
 // Authenticate checks for the presence of a valid token when accessing server-side resources.
 func (server *ArgoCDServer) Authenticate(ctx context.Context) (context.Context, error) {
+	var span trace.Span
+	ctx, span = tracer.Start(ctx, "server.ArgoCDServer.Authenticate")
+	defer span.End()
 	if server.DisableAuth {
 		return ctx, nil
 	}
@@ -1561,18 +1573,25 @@ func (server *ArgoCDServer) Authenticate(ctx context.Context) (context.Context,
 
 // getClaims extracts, validates and refreshes a JWT token from an incoming request context.
 func (server *ArgoCDServer) getClaims(ctx context.Context) (jwt.Claims, string, error) {
+	var span trace.Span
+	ctx, span = tracer.Start(ctx, "server.ArgoCDServer.getClaims")
+	defer span.End()
 	md, ok := metadata.FromIncomingContext(ctx)
 	if !ok {
+		span.SetStatus(otel_codes.Error, ErrNoSession.Error())
 		return nil, "", ErrNoSession
 	}
 	tokenString := getToken(md)
 	if tokenString == "" {
+		span.SetStatus(otel_codes.Error, ErrNoSession.Error())
 		return nil, "", ErrNoSession
 	}
 	// A valid argocd-issued token is automatically refreshed here prior to expiration.
 	// OIDC tokens will be verified but will not be refreshed here.
 	claims, newToken, err := server.sessionMgr.VerifyToken(ctx, tokenString)
 	if err != nil {
+		span.SetAttributes(attribute.String("token", tokenString))
+		span.SetStatus(otel_codes.Error, err.Error())
 		return claims, "", status.Errorf(codes.Unauthenticated, "invalid session: %v", err)
 	}
 

util/oidc/oidc.go

@@ -18,6 +18,12 @@ import (
 	"sync"
 	"time"
 
+	"go.opentelemetry.io/otel/codes"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
+
 	gooidc "github.com/coreos/go-oidc/v3/oidc"
 	"github.com/golang-jwt/jwt/v5"
 	log "github.com/sirupsen/logrus"
@@ -37,6 +43,13 @@ import (
 
 var ErrInvalidRedirectURL = errors.New("invalid return URL")
 
+// OpenTelemetry tracer for this package
+var tracer trace.Tracer
+
+func init() {
+	tracer = otel.Tracer("github.com/argoproj/argo-cd/v3/util/oidc")
+}
+
 const (
 	GrantTypeAuthorizationCode  = "authorization_code"
 	GrantTypeImplicit           = "implicit"
@@ -146,14 +159,16 @@ func GetOidcTokenCacheFromJSON(jsonBytes []byte) (*OidcTokenCache, error) {
 // GetTokenSourceFromCache creates an oauth2 TokenSource from a cached oidc token.  The TokenSource will be configured
 // with an early expiration based on the refreshTokenThreshold.
 func (a *ClientApp) GetTokenSourceFromCache(ctx context.Context, oidcTokenCache *OidcTokenCache) (oauth2.TokenSource, error) {
+	spanCtx, span := tracer.Start(ctx, "oidc.ClientApp.GetTokenSourceFromCache")
+	defer span.End()
 	if oidcTokenCache == nil {
 		return nil, errors.New("oidcTokenCache is required")
 	}
 	config, err := a.getOauth2ConfigForRedirectURI(oidcTokenCache.RedirectURL)
 	if err != nil {
 		return nil, err
 	}
-	baseTokenSource := config.TokenSource(ctx, oidcTokenCache.Token)
+	baseTokenSource := config.TokenSource(spanCtx, oidcTokenCache.Token)
 	tokenRefresher := oauth2.ReuseTokenSourceWithExpiry(oidcTokenCache.Token, baseTokenSource, a.refreshTokenThreshold)
 	return tokenRefresher, nil
 }
@@ -197,10 +212,18 @@ func NewClientApp(settings *settings.ArgoCDSettings, dexServerAddr string, dexTL
 
 	transport := &http.Transport{
 		Proxy: http.ProxyFromEnvironment,
-		Dial: (&net.Dialer{
-			Timeout:   30 * time.Second,
-			KeepAlive: 30 * time.Second,
-		}).Dial,
+		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			_, span := tracer.Start(ctx, "oidc.ClientApp.client")
+			span.SetAttributes(
+				attribute.String("network", network),
+				attribute.String("addr", addr),
+			)
+			defer span.End()
+			return (&net.Dialer{
+				Timeout:   30 * time.Second,
+				KeepAlive: 30 * time.Second,
+			}).Dial(network, addr)
+		},
 		TLSHandshakeTimeout:   10 * time.Second,
 		ExpectContinueTimeout: 1 * time.Second,
 	}
@@ -540,7 +563,7 @@ func (a *ClientApp) HandleCallback(w http.ResponseWriter, r *http.Request) {
 	}
 	// save the accessToken in memory for later use
 	sub := jwtutil.StringField(claims, "sub")
-	err = a.SetValueInEncryptedCache(FormatAccessTokenCacheKey(sub), []byte(token.AccessToken), GetTokenExpiration(claims))
+	err = a.SetValueInEncryptedCache(ctx, FormatAccessTokenCacheKey(sub), []byte(token.AccessToken), GetTokenExpiration(claims))
 	if err != nil {
 		claimsJSON, _ := json.Marshal(claims)
 		log.Errorf("cannot cache encrypted accessToken: %v (claims=%s)", err, claimsJSON)
@@ -556,7 +579,7 @@ func (a *ClientApp) HandleCallback(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	sid := jwtutil.StringField(claims, "sid")
-	err = a.SetValueInEncryptedCache(formatOidcTokenCacheKey(sub, sid), oidcTokenCacheJSON, GetTokenExpiration(claims))
+	err = a.SetValueInEncryptedCache(ctx, formatOidcTokenCacheKey(sub, sid), oidcTokenCacheJSON, GetTokenExpiration(claims))
 	if err != nil {
 		claimsJSON, _ := json.Marshal(claims)
 		log.Errorf("cannot cache encrypted oidc token: %v (claims=%s)", err, claimsJSON)
@@ -586,56 +609,87 @@ func (a *ClientApp) HandleCallback(w http.ResponseWriter, r *http.Request) {
 // GetValueFromEncryptedCache is a convenience method for retreiving a value from cache and decrypting it.  If the cache
 // does not contain a value for the given key, a nil value is returned.  Return handling should check for error and then
 // check for nil.
-func (a *ClientApp) GetValueFromEncryptedCache(key string) (value []byte, err error) {
+func (a *ClientApp) GetValueFromEncryptedCache(ctx context.Context, key string) (value []byte, err error) {
+	_, span := tracer.Start(ctx, "oidc.ClientApp.GetValueFromEncryptedCache")
+	defer span.End()
 	var encryptedValue []byte
+	span.AddEvent("start cache read")
 	err = a.clientCache.Get(key, &encryptedValue)
+	span.AddEvent("end cache read")
 	if err != nil {
 		if errors.Is(err, cache.ErrCacheMiss) {
+			span.SetAttributes(attribute.Bool("cache_hit", false))
 			// Return nil to signify a cache miss
 			return nil, nil
 		}
-		return nil, fmt.Errorf("failed to get encrypted value from cache: %w", err)
+		err = fmt.Errorf("failed to get encrypted value from cache: %w", err)
+		span.SetStatus(codes.Error, err.Error())
+		return nil, err
 	}
+	span.SetAttributes(attribute.Bool("cache_hit", true))
 	value, err = crypto.Decrypt(encryptedValue, a.encryptionKey)
 	if err != nil {
-		return nil, fmt.Errorf("failed to decrypt value from cache: %w", err)
+		err = fmt.Errorf("failed to decrypt value from cache: %w", err)
+		span.SetStatus(codes.Error, err.Error())
+		return nil, err
 	}
 	return value, err
 }
 
 // SetValueFromEncyrptedCache is a convenience method for encrypting a value and storing it in the cache at a given key.
 // Cache expiration is set based on input.
-func (a *ClientApp) SetValueInEncryptedCache(key string, value []byte, expiration time.Duration) error {
+func (a *ClientApp) SetValueInEncryptedCache(ctx context.Context, key string, value []byte, expiration time.Duration) error {
+	_, span := tracer.Start(ctx, "oidc.ClientApp.SetValueInEncryptedCache")
+	defer span.End()
 	encryptedValue, err := crypto.Encrypt(value, a.encryptionKey)
 	if err != nil {
+		span.SetStatus(codes.Error, err.Error())
 		return err
 	}
+	span.SetAttributes(
+		attribute.String("key", key),
+		attribute.Int("value_length", len(value)),
+	)
+	span.AddEvent("start cache write")
 	err = a.clientCache.Set(&cache.Item{
 		Key:    key,
 		Object: encryptedValue,
 		CacheActionOpts: cache.CacheActionOpts{
 			Expiration: expiration,
 		},
 	})
+	span.AddEvent("end cache write")
 	if err != nil {
+		span.SetStatus(codes.Error, err.Error())
 		return err
 	}
 	return nil
 }
 
 func (a *ClientApp) CheckAndRefreshToken(ctx context.Context, groupClaims jwt.MapClaims, refreshTokenThreshold time.Duration) (string, error) {
+	var span trace.Span
+	ctx, span = tracer.Start(ctx, "oidc.ClientApp.CheckAndRefreshToken")
+	defer span.End()
+	iss := jwtutil.StringField(groupClaims, "iss")
 	sub := jwtutil.StringField(groupClaims, "sub")
 	sid := jwtutil.StringField(groupClaims, "sid")
+	span.SetAttributes(
+		attribute.String("iss", iss),
+		attribute.String("sub", sub),
+		attribute.String("sid", sid))
 	if GetTokenExpiration(groupClaims) < refreshTokenThreshold {
 		token, err := a.GetUpdatedOidcTokenFromCache(ctx, sub, sid)
 		if err != nil {
 			log.Errorf("Failed to get token from cache: %v", err)
+			span.SetStatus(codes.Error, err.Error())
 			return "", err
 		}
 		if token != nil {
 			idTokenRAW, ok := token.Extra("id_token").(string)
 			if !ok {
-				return "", errors.New("empty id_token")
+				err = errors.New("empty id_token")
+				span.SetStatus(codes.Error, err.Error())
+				return "", err
 			}
 			return idTokenRAW, nil
 		}
@@ -646,12 +700,23 @@ func (a *ClientApp) CheckAndRefreshToken(ctx context.Context, groupClaims jwt.Ma
 // GetUpdatedOidcTokenFromCache fetches a token from cache and refreshes it if under the threshold for expiration.
 // The cached token will also be updated if it is refreshed.  Returns latest token or an error if the process fails.
 func (a *ClientApp) GetUpdatedOidcTokenFromCache(ctx context.Context, subject string, sessionId string) (*oauth2.Token, error) {
+	var span trace.Span
+	ctx, span = tracer.Start(ctx, "oidc.ClientApp.GetUpdatedOidcTokenFromCache")
+	defer span.End()
+
 	ctx = gooidc.ClientContext(ctx, a.client)
+	if span.IsRecording() {
+		span.SetAttributes(
+			attribute.String("subject", subject),
+			attribute.String("sessionId", sessionId),
+		)
+	}
 
 	// Get oauth2 config
 	cacheKey := formatOidcTokenCacheKey(subject, sessionId)
-	oidcTokenCacheJSON, err := a.GetValueFromEncryptedCache(cacheKey)
+	oidcTokenCacheJSON, err := a.GetValueFromEncryptedCache(ctx, cacheKey)
 	if err != nil {
+		span.SetStatus(codes.Error, err.Error())
 		return nil, err
 	}
 	if oidcTokenCacheJSON == nil {
@@ -661,25 +726,35 @@ func (a *ClientApp) GetUpdatedOidcTokenFromCache(ctx context.Context, subject st
 	oidcTokenCache, err := GetOidcTokenCacheFromJSON(oidcTokenCacheJSON)
 	if err != nil {
 		err = fmt.Errorf("failed to unmarshal cached oidc token: %w", err)
+		span.SetStatus(codes.Error, err.Error())
 		return nil, err
 	}
 	tokenSource, err := a.GetTokenSourceFromCache(ctx, oidcTokenCache)
 	if err != nil {
 		err = fmt.Errorf("failed to get token source from cached oidc token: %w", err)
+		span.SetStatus(codes.Error, err.Error())
 		return nil, err
 	}
+	span.AddEvent("starting tokenSource.Token()")
 	token, err := tokenSource.Token()
+	span.AddEvent("finished tokenSource.Token()")
 	if err != nil {
-		return nil, fmt.Errorf("failed to refresh token from source: %w", err)
+		err = fmt.Errorf("failed to refresh token from source: %w", err)
+		span.SetStatus(codes.Error, err.Error())
+		return nil, err
 	}
 	if token.AccessToken != oidcTokenCache.Token.AccessToken {
+		span.AddEvent("updating cache with latest token")
 		oidcTokenCache = NewOidcTokenCache(oidcTokenCache.RedirectURL, token)
 		oidcTokenCacheJSON, err = json.Marshal(oidcTokenCache)
 		if err != nil {
-			return nil, fmt.Errorf("failed to marshal oidc oidcTokenCache refresher: %w", err)
+			err = fmt.Errorf("failed to marshal oidc oidcTokenCache refresher: %w", err)
+			span.SetStatus(codes.Error, err.Error())
+			return nil, err
 		}
-		err = a.SetValueInEncryptedCache(cacheKey, oidcTokenCacheJSON, time.Until(token.Expiry))
+		err = a.SetValueInEncryptedCache(ctx, cacheKey, oidcTokenCacheJSON, time.Until(token.Expiry))
 		if err != nil {
+			span.SetStatus(codes.Error, err.Error())
 			return nil, err
 		}
 	}
@@ -833,6 +908,9 @@ func (a *ClientApp) SetGroupsFromUserInfo(ctx context.Context, claims jwt.Claims
 
 // GetUserInfo queries the IDP userinfo endpoint for claims
 func (a *ClientApp) GetUserInfo(ctx context.Context, actualClaims jwt.MapClaims, issuerURL, userInfoPath string) (jwt.MapClaims, bool, error) {
+	var span trace.Span
+	ctx, span = tracer.Start(ctx, "oidc.ClientApp.GetUserInfo")
+	defer span.End()
 	sub := jwtutil.StringField(actualClaims, "sub")
 	var claims jwt.MapClaims
 	var encClaims []byte
@@ -854,7 +932,7 @@ func (a *ClientApp) GetUserInfo(ctx context.Context, actualClaims jwt.MapClaims,
 	}
 
 	// check if the accessToken for the user is still present
-	accessTokenBytes, err := a.GetValueFromEncryptedCache(FormatAccessTokenCacheKey(sub))
+	accessTokenBytes, err := a.GetValueFromEncryptedCache(ctx, FormatAccessTokenCacheKey(sub))
 	if err != nil {
 		return claims, true, fmt.Errorf("could not read accessToken from cache for %s: %w", sub, err)
 	}