Ship traces to opensearch

apis/v1beta2/dataprepper_types.go

@@ -34,13 +34,14 @@ type DataPrepperSpec struct {
 	// +kubebuilder:default:=latest
 	Version string `json:"version"`
 	// +optional
-	DefaultRepo  *string                   `json:"defaultRepo,omitempty"`
-	Opensearch   *OpensearchSpec           `json:"opensearch,omitempty"`
-	Username     string                    `json:"username"`
-	PasswordFrom *corev1.SecretKeySelector `json:"passwordFrom,omitempty"`
-	ClusterID    string                    `json:"cluster,omitempty"`
-	NodeSelector map[string]string         `json:"nodeSelector,omitempty"`
-	Tolerations  []corev1.Toleration       `json:"tolerations,omitempty"`
+	DefaultRepo   *string                   `json:"defaultRepo,omitempty"`
+	Opensearch    *OpensearchSpec           `json:"opensearch,omitempty"`
+	Username      string                    `json:"username"`
+	PasswordFrom  *corev1.SecretKeySelector `json:"passwordFrom,omitempty"`
+	ClusterID     string                    `json:"cluster,omitempty"`
+	NodeSelector  map[string]string         `json:"nodeSelector,omitempty"`
+	Tolerations   []corev1.Toleration       `json:"tolerations,omitempty"`
+	EnableTracing bool                      `json:"enableTracing,omitempty"`
 }
 
 type OpensearchSpec struct {

config/agent-crd/bases/opni.io_datapreppers.yaml

@@ -35,6 +35,8 @@ spec:
                 type: string
               defaultRepo:
                 type: string
+              enableTracing:
+                type: boolean
               image:
                 type: string
               imagePullPolicy:

config/crd/bases/opni.io_datapreppers.yaml

@@ -35,6 +35,8 @@ spec:
                 type: string
               defaultRepo:
                 type: string
+              enableTracing:
+                type: boolean
               image:
                 type: string
               imagePullPolicy:

go.mod

@@ -20,7 +20,7 @@ require (
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/cert-manager/cert-manager v1.8.0
 	github.com/cortexproject/cortex v1.13.0
-	github.com/dbason/featureflags v0.0.0-20220731224053-be63751dd521
+	github.com/dbason/featureflags v0.0.0-20220803034705-b6242a8d72b2
 	github.com/gin-gonic/gin v1.8.1
 	github.com/go-kit/log v0.2.0
 	github.com/go-logr/logr v1.2.3
@@ -120,6 +120,7 @@ require (
 	k8s.io/apimachinery v0.24.3
 	k8s.io/client-go v0.24.3
 	k8s.io/component-base v0.24.3
+	k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9
 	opensearch.opster.io v0.0.0-00010101000000-000000000000
 	sigs.k8s.io/controller-runtime v0.11.1
 	sigs.k8s.io/controller-tools v0.8.1-0.20220428122951-32ad71090a62
@@ -461,7 +462,6 @@ require (
 	k8s.io/klog/v2 v2.60.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20220328201542-3ee0da9b0b42 // indirect
 	k8s.io/kubectl v0.24.3 // indirect
-	k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9 // indirect
 	lukechampine.com/frand v1.4.2 // indirect
 	oras.land/oras-go v1.1.0 // indirect
 	rsc.io/binaryregexp v0.2.0 // indirect

go.sum

@@ -633,8 +633,8 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/daviddengcn/go-colortext v0.0.0-20160507010035-511bcaf42ccd/go.mod h1:dv4zxwHi5C/8AeI+4gX4dCWOIvNi7I6JCSX0HvlKPgE=
 github.com/dbason/charts-build-scripts v0.3.4-0.20220429024555-807c076e8116 h1:huvFnqlUpiIldOsQoM+Ad6ZE/oJfyzyB9A8W2yt4hLc=
 github.com/dbason/charts-build-scripts v0.3.4-0.20220429024555-807c076e8116/go.mod h1:pzDiuRy1QgnVI8yq1e3d7WvuofNAzgj/c48IRNX2n38=
-github.com/dbason/featureflags v0.0.0-20220731224053-be63751dd521 h1:pXGegKRlm11AL1IC7PLdkNRpQuGLa8TJSQIuE7jltVU=
-github.com/dbason/featureflags v0.0.0-20220731224053-be63751dd521/go.mod h1:ew+9wAV4nyz0ZKybKcM4cMosZwggyAkzU3BU/wZxHTA=
+github.com/dbason/featureflags v0.0.0-20220803034705-b6242a8d72b2 h1:GPBUVhDuyLGWicbQ0FPVF1ZxCOI/0rJ7C3b64nN16oY=
+github.com/dbason/featureflags v0.0.0-20220803034705-b6242a8d72b2/go.mod h1:ew+9wAV4nyz0ZKybKcM4cMosZwggyAkzU3BU/wZxHTA=
 github.com/dbason/opensearch-k8s-operator/opensearch-operator v0.0.0-20220711232325-352be96017bf h1:KjLbelOgWjJzn9h/PV8WAWN8APDW08eUstdabeKk4Q4=
 github.com/dbason/opensearch-k8s-operator/opensearch-operator v0.0.0-20220711232325-352be96017bf/go.mod h1:pRMCDq01cGnDy3x8SszLoZhneMvxJor9JFqc8U1YE/0=
 github.com/decred/dcrd/crypto/blake256 v1.0.0/go.mod h1:sQl2p6Y26YV+ZOcSTP6thNdn47hh8kt6rqSlvmrXFAc=

packages/opni-agent/opni-agent/charts/crds/crds.yaml

@@ -11353,6 +11353,8 @@ spec:
                 type: string
               defaultRepo:
                 type: string
+              enableTracing:
+                type: boolean
               image:
                 type: string
               imagePullPolicy:

packages/opni-agent/opni-agent/charts/templates/job.yaml

@@ -41,6 +41,9 @@ spec:
         {{- if .Values.logging.disableOpensearchTLSVerify }}
         - --opensearch-insecure
         {{- end }}
+        {{- if .Values.logging.tracing }}
+        - --tracing
+        {{- end }}
       restartPolicy: Never
   backoffLimit: 3
 {{- end }}

packages/opni-agent/opni-agent/charts/values.yaml

@@ -18,6 +18,7 @@ metrics:
 logging:
   enabled: false
   insecure: false
+  tracing: false
   disableOpensearchTLSVerify: false
   disableBootstrap: false
   provider: rke

packages/opni/opni/charts/crds/crds.yaml

@@ -47,6 +47,8 @@ spec:
                   leaseID:
                     format: int64
                     type: integer
+                  resourceVersion:
+                    type: string
                   ttl:
                     format: int64
                     type: integer
@@ -14971,6 +14973,15 @@ spec:
                   capabilities:
                     items:
                       properties:
+                        deletionTimestamp:
+                          properties:
+                            nanos:
+                              format: int32
+                              type: integer
+                            seconds:
+                              format: int64
+                              type: integer
+                          type: object
                         name:
                           type: string
                       type: object
@@ -14979,6 +14990,8 @@ spec:
                     additionalProperties:
                       type: string
                     type: object
+                  resourceVersion:
+                    type: string
                 type: object
             type: object
         type: object
@@ -15021,6 +15034,8 @@ spec:
                 type: string
               defaultRepo:
                 type: string
+              enableTracing:
+                type: boolean
               image:
                 type: string
               imagePullPolicy:

packages/opni/opni/charts/templates/features.yaml

@@ -7,4 +7,7 @@ data:
   features: |
     manage-opensearch:
       description: Management API for Opensearch Clusters
+      enabled: false
+    tracing:
+      description: Trace collection feature
       enabled: false

pkg/opni/commands/agent.go

@@ -7,8 +7,8 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
-	"sync"
 	"syscall"
+	"time"
 
 	"github.com/go-logr/zapr"
 	upgraderesponder "github.com/longhorn/upgrade-responder/client"
@@ -52,6 +52,57 @@ var (
 func BuildAgentCmd() *cobra.Command {
 	var disableUsage bool
 
+	run := func(parentCtx context.Context) error {
+		ctx := waitctx.FromContext(parentCtx)
+
+		tracing.Configure("agent")
+		agentlg = logger.New(logger.WithLogLevel(util.Must(zapcore.ParseLevel(agentLogLevel))))
+
+		if os.Getenv("DO_NOT_TRACK") == "1" {
+			disableUsage = true
+		}
+
+		var upgradeChecker *upgraderesponder.UpgradeChecker
+		if !(disableUsage || common.DisableUsage) {
+			upgradeRequester := manager.UpgradeRequester{
+				Version:     util.Version,
+				InstallType: manager.InstallTypeAgent,
+			}
+			upgradeRequester.SetupLogger(zapr.NewLogger(agentlg.Desugar()))
+			setupLog.Info("Usage tracking enabled", "current-version", util.Version)
+			upgradeChecker = upgraderesponder.NewUpgradeChecker(upgradeResponderAddress, &upgradeRequester)
+			upgradeChecker.Start()
+			defer upgradeChecker.Stop()
+		}
+
+		if enableMetrics {
+			waitctx.Go(ctx, func() {
+				runMonitoringAgent(ctx)
+			})
+		}
+
+		if enableLogging {
+			waitctx.Go(ctx, func() {
+				err := runLoggingControllers(ctx)
+				if err != nil {
+					agentlg.Fatalf("failed to start controllers: %v", err)
+				}
+			})
+		}
+
+		if enableEventCollector {
+			waitctx.Go(ctx, func() {
+				err := runEventsCollector(ctx)
+				if err != nil {
+					agentlg.Fatalf("failed to run event collector: %v", err)
+				}
+			})
+		}
+
+		waitctx.WaitWithTimeout(ctx, 60*time.Second, 10*time.Second)
+		return nil
+	}
+
 	agentCmd := &cobra.Command{
 		Use:   "agent",
 		Short: "Run the Opni Monitoring Agent",
@@ -64,58 +115,7 @@ agent remote-write requests to add dynamic authentication.`,
 			return nil
 		},
 		Run: func(cmd *cobra.Command, args []string) {
-			ctx := waitctx.FromContext(cmd.Context())
-			tracing.Configure("agent")
-			agentlg = logger.New(logger.WithLogLevel(util.Must(zapcore.ParseLevel(agentLogLevel))))
-			wg := sync.WaitGroup{}
-
-			if os.Getenv("DO_NOT_TRACK") == "1" {
-				disableUsage = true
-			}
-
-			if !(disableUsage || common.DisableUsage) {
-				upgradeRequester := manager.UpgradeRequester{
-					Version:     util.Version,
-					InstallType: manager.InstallTypeAgent,
-				}
-				upgradeRequester.SetupLogger(zapr.NewLogger(agentlg.Desugar()))
-				setupLog.Info("Usage tracking enabled", "current-version", util.Version)
-				upgradeChecker := upgraderesponder.NewUpgradeChecker(upgradeResponderAddress, &upgradeRequester)
-				upgradeChecker.Start()
-				defer upgradeChecker.Stop()
-			}
-
-			if enableMetrics {
-				wg.Add(1)
-				go func(ctx context.Context) {
-					defer wg.Done()
-					runMonitoringAgent(ctx)
-				}(ctx)
-			}
-
-			if enableLogging {
-				wg.Add(1)
-				go func(ctx context.Context) {
-					defer wg.Done()
-					err := runLoggingControllers(ctx)
-					if err != nil {
-						agentlg.Fatalf("failed to start controllers: %v", err)
-					}
-				}(ctx)
-			}
-
-			if enableEventCollector {
-				wg.Add(1)
-				go func(ctx context.Context) {
-					defer wg.Done()
-					err := runEventsCollector(ctx)
-					if err != nil {
-						agentlg.Fatalf("failed to run event collector: %v", err)
-					}
-				}(cmd.Context())
-			}
-
-			wg.Wait()
+			run(cmd.Context())
 		},
 	}
 
@@ -319,7 +319,8 @@ func runLoggingControllers(ctx context.Context) error {
 		}
 	}
 
-	ctx, _ = signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
+	ctx, cancel := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
+	defer cancel()
 
 	setupLog.Info("starting manager")
 	if err := mgr.Start(ctx); err != nil {

pkg/opni/commands/bootstrap_logging.go

@@ -41,6 +41,7 @@ var (
 	provider        string
 	namespace       string
 	pins            []string
+	enableTracing   bool
 
 	k8sClient client.Client
 )
@@ -60,6 +61,7 @@ func BuildBootstrapLoggingCmd() *cobra.Command {
 	command.Flags().BoolVar(&skipTLSVerify, "opensearch-insecure", false, "skip Opensearch tls verification")
 	command.Flags().BoolVar(&rancherLogging, "use-rancher-logging", false, "manually configure log shipping with rancher-logging")
 	command.Flags().BoolVar(&inCluster, "in-cluster", false, "set bootstrap to run in cluster")
+	command.Flags().BoolVar(&enableTracing, "tracing", false, "Enable OTEL trace shipping")
 	command.Flags().StringVar(&gatewayEndpoint, "gateway-url", "https://localhost:8443", "upstream Opni gateway")
 	command.Flags().StringVar(&provider, "provider", "rke", "the Kubernetes distribution")
 	command.Flags().StringVar(&bootstrapToken, "token", "", "bootstrap token")
@@ -126,6 +128,7 @@ func doBootstrap(cmd *cobra.Command, args []string) error {
 		creds.Username,
 		clusterID,
 		creds.ExternalURL,
+		creds.TracingEnabled,
 	); err != nil {
 		return err
 	}
@@ -164,6 +167,7 @@ func createDataPrepper(
 	username string,
 	clusterID string,
 	opensearchEndpoint string,
+	tracingFeature bool,
 ) error {
 	dataPrepper := v1beta2.DataPrepper{
 		ObjectMeta: metav1.ObjectMeta{
@@ -183,6 +187,12 @@ func createDataPrepper(
 				InsecureDisableSSLVerify: skipTLSVerify,
 			},
 			ClusterID: clusterID,
+			EnableTracing: func() bool {
+				if !tracingFeature {
+					return false
+				}
+				return enableTracing
+			}(),
 		},
 	}
 

pkg/opni/commands/gateway.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"sync/atomic"
+	"time"
 
 	"github.com/hashicorp/go-plugin"
 	"github.com/rancher/opni/pkg/config"
@@ -165,7 +166,7 @@ func BuildGatewayCmd() *cobra.Command {
 		lg.Info(style.Style("waiting for servers to shut down"))
 		fCancel()
 		cancel()
-		waitctx.Wait(ctx)
+		waitctx.WaitWithTimeout(ctx, 60*time.Second, 10*time.Second)
 
 		atomic.StoreUint32(&plugin.Killed, 0)
 		lg.Info(style.Style("--- reloading ---"))
@@ -176,6 +177,7 @@ func BuildGatewayCmd() *cobra.Command {
 		Use:   "gateway",
 		Short: "Run the Opni Monitoring Gateway",
 		RunE: func(cmd *cobra.Command, args []string) error {
+			defer waitctx.RecoverTimeout()
 			for {
 				if err := run(); err != nil {
 					return err