diff --git a/aggregator/events.go b/aggregator/events.go index 30930b827..b15be067c 100644 --- a/aggregator/events.go +++ b/aggregator/events.go @@ -4,6 +4,7 @@ import ( "app/base/database" "app/base/mqueue" "app/base/utils" + "context" "sync" "time" @@ -43,7 +44,7 @@ func flushAdvisoryBuffer() { processAdvisoryBatch(grouped) } -func advisoryUpdateHandler(m mqueue.KafkaMessage) error { +func advisoryUpdateHandler(_ context.Context, m mqueue.KafkaMessage) error { var event mqueue.AdvisoryUpdateEvent if err := sonic.Unmarshal(m.Value, &event); err != nil { utils.LogError("err", err, "could not deserialize advisory update event") diff --git a/aggregator/events_test.go b/aggregator/events_test.go index 692c98fa0..b07d7d98c 100644 --- a/aggregator/events_test.go +++ b/aggregator/events_test.go @@ -5,6 +5,7 @@ import ( "app/base/database" "app/base/mqueue" "app/base/utils" + "context" "sort" "testing" "time" @@ -75,13 +76,13 @@ func TestBufferedEventsProcessedOnBatchThreshold(t *testing.T) { msg := toKafkaMessage(t, mqueue.AdvisoryUpdateEvent{RhAccountID: 1, AdvisoryIDs: []int64{1, 2}}) // First two events accumulate in the buffer - assert.Nil(t, advisoryUpdateHandler(msg)) + assert.Nil(t, advisoryUpdateHandler(context.Background(), msg)) assert.Equal(t, 1, len(advisoryBuffer)) - assert.Nil(t, advisoryUpdateHandler(msg)) + assert.Nil(t, advisoryUpdateHandler(context.Background(), msg)) assert.Equal(t, 2, len(advisoryBuffer)) // Third event triggers flush and processAdvisoryBatch runs - assert.Nil(t, advisoryUpdateHandler(msg)) + assert.Nil(t, advisoryUpdateHandler(context.Background(), msg)) assert.Equal(t, 0, len(advisoryBuffer)) // Verify account_advisory was populated diff --git a/base/api/client.go b/base/api/client.go index 1200f8932..3983e08c4 100644 --- a/base/api/client.go +++ b/base/api/client.go @@ -1,11 +1,13 @@ package api import ( + "app/base/telemetry" "app/base/utils" "bytes" "context" "io" "net/http" + "sync" "github.com/bytedance/sonic" "github.com/pkg/errors" @@ -15,10 +17,15 @@ type Client struct { HTTPClient *http.Client Debug bool DefaultHeaders map[string]string + otelOnce sync.Once } func (o *Client) Request(ctx *context.Context, method, url string, requestPtr interface{}, responseOutPtr interface{}) (*http.Response, error) { + o.otelOnce.Do(func() { + o.HTTPClient = telemetry.InstrumentHTTPClient(o.HTTPClient) + }) + body := &bytes.Buffer{} if requestPtr != nil { err := sonic.ConfigDefault.NewEncoder(body).Encode(requestPtr) diff --git a/base/base.go b/base/base.go index 222b1c9c1..344e719fc 100644 --- a/base/base.go +++ b/base/base.go @@ -1,6 +1,7 @@ package base import ( + "app/base/telemetry" "app/base/utils" "context" "os" @@ -31,6 +32,7 @@ func HandleSignals() { utils.LogInfo("starting grace period for " + sig.String()) time.Sleep(defaultK8sGracePeriod / 4) CancelContext() + _ = telemetry.Shutdown(context.Background()) utils.LogInfo("SIGTERM/SIGINT handled") }() } diff --git a/base/core/config.go b/base/core/config.go index 42506ec4f..1ef0c4c42 100644 --- a/base/core/config.go +++ b/base/core/config.go @@ -3,6 +3,7 @@ package core import ( "app/base/database" "app/base/metrics" + "app/base/telemetry" "app/base/utils" "testing" ) @@ -14,18 +15,26 @@ var ( dbWait = utils.PodConfig.GetString("wait_for_db", "empty") ) -func configureBaseApp() { +func initObservability() { utils.ConfigureLogging() + if err := telemetry.Init(); err != nil { + panic(err) + } +} + +func configureBaseApp() { metrics.Configure() database.DBWait(dbWait) } func ConfigureApp() { + initObservability() database.Configure() configureBaseApp() } func ConfigureAdminApp() { + initObservability() database.ConfigureAdmin() configureBaseApp() } diff --git a/base/database/setup.go b/base/database/setup.go index 1d635d7ec..8ab1ce475 100644 --- a/base/database/setup.go +++ b/base/database/setup.go @@ -1,10 +1,13 @@ package database import ( + "app/base/telemetry" "app/base/utils" "fmt" "time" + "github.com/XSAM/otelsql" + semconv "go.opentelemetry.io/otel/semconv/v1.24.0" "gorm.io/driver/postgres" "gorm.io/gorm" "gorm.io/gorm/logger" @@ -88,7 +91,21 @@ func createGormConfig(debug bool) *gorm.Config { // open database connection func openPostgreSQL(dbConfig *PostgreSQLConfig) *gorm.DB { connectString := dataSourceName(dbConfig) - db, err := gorm.Open(postgres.Open(connectString), createGormConfig(dbConfig.Debug)) + cfg := createGormConfig(dbConfig.Debug) + + var db *gorm.DB + var err error + if telemetry.SQLEnabled() { + // gorm.io/driver/postgres uses jackc/pgx/v5/stdlib (driver name "pgx") + sqlDB, openErr := otelsql.Open("pgx", connectString, + otelsql.WithAttributes(semconv.DBSystemPostgreSQL)) + if openErr != nil { + panic(openErr) + } + db, err = gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), cfg) + } else { + db, err = gorm.Open(postgres.Open(connectString), cfg) + } if err != nil { panic(err) } diff --git a/base/mqueue/mqueue.go b/base/mqueue/mqueue.go index 7c182a662..06d2be990 100644 --- a/base/mqueue/mqueue.go +++ b/base/mqueue/mqueue.go @@ -54,18 +54,18 @@ type KafkaMessage struct { Headers []kafka.Header } -type MessageHandler func(message KafkaMessage) error +type MessageHandler func(ctx context.Context, message KafkaMessage) error func MakeRetryingHandler(handler MessageHandler) MessageHandler { - return func(message KafkaMessage) error { + return func(ctx context.Context, message KafkaMessage) error { var err error var attempt int - ctx, cancel := context.WithCancel(context.Background()) - backoffState := policy.Start(ctx) + backoffCtx, cancel := context.WithCancel(context.Background()) + backoffState := policy.Start(backoffCtx) defer cancel() for backoff.Continue(backoffState) { - if err = handler(message); err == nil || !errors.Is(err, base.ErrFatal) { + if err = handler(ctx, message); err == nil || !errors.Is(err, base.ErrFatal) { return nil } utils.LogError("err", err, "attempt", attempt, "Try failed") diff --git a/base/mqueue/mqueue_impl_gokafka.go b/base/mqueue/mqueue_impl_gokafka.go index d29eb6f3e..2f1c8936c 100644 --- a/base/mqueue/mqueue_impl_gokafka.go +++ b/base/mqueue/mqueue_impl_gokafka.go @@ -1,6 +1,7 @@ package mqueue import ( + "app/base/telemetry" "app/base/utils" "context" "crypto/tls" @@ -41,7 +42,10 @@ func (t *kafkaGoReaderImpl) HandleMessages(ctx context.Context, handler MessageH } // At this level, all errors are fatal kafkaMessage := KafkaMessage{Key: m.Key, Value: m.Value, Headers: m.Headers} - if err = handler(kafkaMessage); err != nil { + msgCtx, span := telemetry.ConsumerContext(ctx, t.Config().Topic, m.Headers) + err = handler(msgCtx, kafkaMessage) + telemetry.End(span, err) + if err != nil { utils.LogPanic("err", err, "Handler failed") } err = t.CommitMessages(ctx, m) @@ -62,7 +66,8 @@ type kafkaGoWriterImpl struct { func (t *kafkaGoWriterImpl) WriteMessages(ctx context.Context, msgs ...KafkaMessage) error { kafkaGoMessages := make([]kafka.Message, len(msgs)) for i, m := range msgs { - kafkaGoMessages[i] = kafka.Message{Key: m.Key, Value: m.Value, Headers: m.Headers} + headers := telemetry.Inject(ctx, m.Headers) + kafkaGoMessages[i] = kafka.Message{Key: m.Key, Value: m.Value, Headers: headers} } err := t.Writer.WriteMessages(ctx, kafkaGoMessages...) return err diff --git a/base/mqueue/mqueue_test.go b/base/mqueue/mqueue_test.go index 6f3ae03ec..df94dfc43 100644 --- a/base/mqueue/mqueue_test.go +++ b/base/mqueue/mqueue_test.go @@ -23,7 +23,7 @@ func TestRoundTripKafkaGo(t *testing.T) { defer reader.Close() var eventOut PlatformEvent - go reader.HandleMessages(t.Context(), func(m KafkaMessage) error { + go reader.HandleMessages(t.Context(), func(_ context.Context, m KafkaMessage) error { return sonic.Unmarshal(m.Value, &eventOut) }) @@ -39,14 +39,14 @@ func TestSpawnReader(t *testing.T) { var nReaders int32 wg := sync.WaitGroup{} SpawnReader(context.Background(), &wg, "", CreateCountedMockReader(&nReaders), - func(_ KafkaMessage) error { return nil }) + func(_ context.Context, _ KafkaMessage) error { return nil }) wg.Wait() assert.Equal(t, 1, int(nReaders)) } func TestRetry(t *testing.T) { i := 0 - handler := func(_ KafkaMessage) error { + handler := func(_ context.Context, _ KafkaMessage) error { i++ if i < 2 { return errors.New("Failed") @@ -55,8 +55,14 @@ func TestRetry(t *testing.T) { } // Without retry handler should fail - assert.Error(t, handler(msg)) + assert.Error(t, handler(context.Background(), msg)) // With retry we handler should eventually succeed - assert.NoError(t, MakeRetryingHandler(handler)(msg)) + assert.NoError(t, MakeRetryingHandler(handler)(context.Background(), msg)) +} + +func TestWriteMessagesInjectsTraceparent(t *testing.T) { + // WriteMessages requires a Kafka broker; inject coverage lives in + // telemetry.Inject / TestProducerContextLinksOriginalsAndInjectsOwnTraceparent. + t.Skip("no broker-free WriteMessages path; covered by telemetry.Inject tests") } diff --git a/base/mqueue/platform_event.go b/base/mqueue/platform_event.go index b138c1358..c457b2ca1 100644 --- a/base/mqueue/platform_event.go +++ b/base/mqueue/platform_event.go @@ -13,15 +13,16 @@ import ( // PlatformEvent ID is typed as uuid.UUID to match the inventory service contract: // https://github.com/RedHatInsights/insights-host-inventory/blob/master/swagger/host_events.spec.yaml type PlatformEvent struct { - ID uuid.UUID `json:"id"` - Type *string `json:"type"` - Timestamp *types.Rfc3339Timestamp `json:"timestamp"` - AccountID int `json:"account_id"` - OrgID *string `json:"org_id,omitempty"` - B64Identity *string `json:"b64_identity"` - URL *string `json:"url"` - SystemIDs []uuid.UUID `json:"system_ids,omitempty"` - RequestIDs []string `json:"request_ids,omitempty"` + ID uuid.UUID `json:"id"` + Type *string `json:"type"` + Timestamp *types.Rfc3339Timestamp `json:"timestamp"` + AccountID int `json:"account_id"` + OrgID *string `json:"org_id,omitempty"` + B64Identity *string `json:"b64_identity"` + URL *string `json:"url"` + SystemIDs []uuid.UUID `json:"system_ids,omitempty"` + RequestIDs []string `json:"request_ids,omitempty"` + Traceparents []string `json:"traceparents,omitempty"` // SkipNotifications suppresses instant advisory notification publish for this event. // Evaluator still marks matching advisory_account_data.notified so later evals do not flood. // Used by recovery recalc; omit/false for normal upload/recalc traffic. @@ -33,6 +34,7 @@ type EvalData struct { RhAccountID int RequestID string OrgID *string + Traceparent string } type PlatformEvents []PlatformEvent @@ -40,6 +42,7 @@ type EvalDataSlice []EvalData type accountInventories map[int][]uuid.UUID type accountRequests map[int][]string +type accountTraceparents map[int][]string type orgIDs map[int]*string func (event *PlatformEvent) createKafkaMessage() (KafkaMessage, error) { @@ -87,19 +90,22 @@ func batchCount(grouped map[int][]uuid.UUID, size int) int { return batches } -func (evals EvalDataSlice) getAccountEvalData(size int) (int, accountInventories, accountRequests, orgIDs) { +func (evals EvalDataSlice) getAccountEvalData(size int) ( + int, accountInventories, accountRequests, accountTraceparents, orgIDs) { // group systems by account invs := accountInventories{} reqs := accountRequests{} + tps := accountTraceparents{} orgs := orgIDs{} for _, e := range evals { invs[e.RhAccountID] = append(invs[e.RhAccountID], e.InventoryID) reqs[e.RhAccountID] = append(reqs[e.RhAccountID], e.RequestID) + tps[e.RhAccountID] = append(tps[e.RhAccountID], e.Traceparent) if _, has := orgs[e.RhAccountID]; !has { orgs[e.RhAccountID] = e.OrgID } } - return batchCount(invs, size), invs, reqs, orgs + return batchCount(invs, size), invs, reqs, tps, orgs } func (evals EvalDataSlice) WriteEvents(ctx context.Context, w Writer) error { @@ -116,7 +122,7 @@ func (evals EvalDataSlice) writeEvents(ctx context.Context, w Writer, size int, if size <= 0 { size = BatchSize } - batches, accInvs, reqs, orgs := evals.getAccountEvalData(size) + batches, accInvs, reqs, tps, orgs := evals.getAccountEvalData(size) now := types.Rfc3339Timestamp(time.Now()) events := make(PlatformEvents, 0, batches) for acc, invs := range accInvs { @@ -130,6 +136,7 @@ func (evals EvalDataSlice) writeEvents(ctx context.Context, w Writer, size int, AccountID: acc, SystemIDs: invs[start:end], RequestIDs: reqs[acc][start:end], + Traceparents: tps[acc][start:end], OrgID: orgs[acc], SkipNotifications: skipNotifications, }) diff --git a/base/mqueue/platform_event_test.go b/base/mqueue/platform_event_test.go index 7f9cf3923..86c6d66ba 100644 --- a/base/mqueue/platform_event_test.go +++ b/base/mqueue/platform_event_test.go @@ -9,8 +9,9 @@ import ( "github.com/stretchr/testify/assert" ) +var orgID string = "org_1" + func TestPlatformEventSkipNotificationsJSON(t *testing.T) { - orgID := "org_1" event := PlatformEvent{ AccountID: 1, OrgID: &orgID, @@ -43,7 +44,6 @@ func TestWriteEventsOfInventoryAccounts(t *testing.T) { var writer Writer = &MockKafkaWriter{} - orgID := "org_1" var invs EvalDataSlice = []EvalData{ {InventoryID: inv2, RhAccountID: acc, OrgID: &orgID}, {InventoryID: inv3, RhAccountID: acc, OrgID: &orgID}} @@ -63,6 +63,25 @@ func TestWriteEventsOfInventoryAccounts(t *testing.T) { assert.False(t, event.SkipNotifications) } +func TestWriteEventsPreservesTraceparents(t *testing.T) { + acc := 1 + inv2 := uuid.MustParse("00000000-0000-0000-0000-000000000002") + inv3 := uuid.MustParse("00000000-0000-0000-0000-000000000003") + writer := &MockKafkaWriter{} + invs := EvalDataSlice{ + {InventoryID: inv2, RhAccountID: acc, OrgID: &orgID, RequestID: "r1", + Traceparent: "00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-bbbbbbbbbbbbbbbb-01"}, + {InventoryID: inv3, RhAccountID: acc, OrgID: &orgID, RequestID: "r2", + Traceparent: "00-cccccccccccccccccccccccccccccccc-dddddddddddddddd-01"}, + } + assert.NoError(t, SendMessages(context.Background(), writer, &invs)) + var event PlatformEvent + assert.NoError(t, sonic.Unmarshal(writer.Messages[0].Value, &event)) + assert.Equal(t, invs[0].Traceparent, event.Traceparents[0]) + assert.Equal(t, invs[1].Traceparent, event.Traceparents[1]) + assert.Equal(t, []string{"r1", "r2"}, event.RequestIDs) +} + func TestWriteEventsSkipNotificationsChunking(t *testing.T) { acc := 7 orgID := "org_recovery" diff --git a/base/telemetry/config.go b/base/telemetry/config.go new file mode 100644 index 000000000..c3e58fcc0 --- /dev/null +++ b/base/telemetry/config.go @@ -0,0 +1,52 @@ +package telemetry + +import ( + "app/base/utils" + "os" + "strconv" + "strings" +) + +const RHServiceName = "patchman-engine" + +func samplingRate() float64 { + raw := os.Getenv("OTEL_SAMPLING_RATE") + if raw == "" { + return 1.0 + } + n, err := strconv.ParseFloat(raw, 64) + if err != nil { + return 1.0 + } + if n < 0 { + return 0 + } + if n > 1 { + return 1 + } + return n +} + +func otelEnabled() bool { + return strings.ToLower(os.Getenv("OTEL_ENABLED")) == "true" +} + +func Enabled() bool { + return enabled +} + +func MQEnabled() bool { + return enabled && utils.GetBoolEnvOrDefault("OTEL_MQ_ENABLED", true) +} + +func HTTPInboundEnabled() bool { + return enabled && utils.GetBoolEnvOrDefault("OTEL_HTTP_INBOUND_ENABLED", true) +} + +func HTTPOutboundEnabled() bool { + return enabled && utils.GetBoolEnvOrDefault("OTEL_HTTP_OUTBOUND_ENABLED", true) +} + +func SQLEnabled() bool { + return enabled && utils.GetBoolEnvOrDefault("OTEL_SQL_ENABLED", true) +} diff --git a/base/telemetry/http.go b/base/telemetry/http.go new file mode 100644 index 000000000..f7dfa8a51 --- /dev/null +++ b/base/telemetry/http.go @@ -0,0 +1,76 @@ +package telemetry + +import ( + "net/http" + + "app/base/utils" + + "github.com/gin-gonic/gin" + "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" +) + +func init() { + // Avoid utils→telemetry import cycle: RunServer calls InstrumentHTTPHandler. + utils.InstrumentHTTPHandler = InstrumentHandler +} + +// TraceHTTPPath reports whether inbound HTTP tracing should create a span for path. +func TraceHTTPPath(path string) bool { + switch path { + case "/healthz", "/livez", "/readyz", "/liveness", "/readiness", "/metrics": + return false + } + if mp := utils.CoreCfg.MetricsPath; mp != "" && path == mp { + return false + } + return true +} + +// InstrumentHandler wraps h with otelhttp when inbound HTTP tracing is enabled. +func InstrumentHandler(h http.Handler) http.Handler { + if !HTTPInboundEnabled() { + return h + } + return otelhttp.NewHandler(h, "http.server", + otelhttp.WithFilter(func(r *http.Request) bool { + return TraceHTTPPath(r.URL.Path) + }), + ) +} + +// InstrumentHTTPClient wraps c.Transport with otelhttp when outbound HTTP tracing is enabled. +func InstrumentHTTPClient(c *http.Client) *http.Client { + if c == nil { + c = &http.Client{} + } + if !HTTPOutboundEnabled() { + return c + } + if _, ok := c.Transport.(*otelhttp.Transport); ok { + return c + } + base := c.Transport + if base == nil { + base = http.DefaultTransport + } + c.Transport = otelhttp.NewTransport(base) + return c +} + +// RHHTTPAttributes sets rh.request_id and rh.org_id on the current span from request headers. +func RHHTTPAttributes() gin.HandlerFunc { + return func(c *gin.Context) { + span := trace.SpanFromContext(c.Request.Context()) + if reqID := c.GetHeader("x-rh-insights-request-id"); reqID != "" { + span.SetAttributes(attribute.String("rh.request_id", reqID)) + } + if identity := c.GetHeader("x-rh-identity"); identity != "" { + if xrhid, err := utils.ParseXRHID(identity); err == nil { + span.SetAttributes(attribute.String("rh.org_id", xrhid.Identity.OrgID)) + } + } + c.Next() + } +} diff --git a/base/telemetry/http_test.go b/base/telemetry/http_test.go new file mode 100644 index 000000000..7448b86be --- /dev/null +++ b/base/telemetry/http_test.go @@ -0,0 +1,17 @@ +package telemetry + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestHTTPFilterExcludesHealthAndMetrics(t *testing.T) { + assert.False(t, TraceHTTPPath("/healthz")) + assert.False(t, TraceHTTPPath("/livez")) + assert.False(t, TraceHTTPPath("/readyz")) + assert.False(t, TraceHTTPPath("/liveness")) + assert.False(t, TraceHTTPPath("/readiness")) + assert.False(t, TraceHTTPPath("/metrics")) + assert.True(t, TraceHTTPPath("/api/patch/v3/advisories")) +} diff --git a/base/telemetry/init.go b/base/telemetry/init.go new file mode 100644 index 000000000..dece4137f --- /dev/null +++ b/base/telemetry/init.go @@ -0,0 +1,164 @@ +package telemetry + +import ( + "app/base/utils" + "context" + "fmt" + "os" + "strings" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + semconv "go.opentelemetry.io/otel/semconv/v1.24.0" + "go.opentelemetry.io/otel/trace" + tracenoop "go.opentelemetry.io/otel/trace/noop" +) + +var ( + enabled bool + tracerProvider *sdktrace.TracerProvider + providerResource *resource.Resource + providerSampler sdktrace.Sampler +) + +func Init() error { + if !otelEnabled() { + otel.SetTracerProvider(tracenoop.NewTracerProvider()) + enabled = false + utils.LogInfo("OpenTelemetry is disabled") + return nil + } + + exp, err := otlptracehttp.New(context.Background(), otlpCompressionOptions()...) + if err != nil { + return err + } + + return initTracerProvider(exp, true) +} + +func initWithExporter(exp sdktrace.SpanExporter) error { + return initTracerProvider(exp, false) +} + +func initTracerProvider(exp sdktrace.SpanExporter, useBatch bool) error { + res, err := newResource() + if err != nil { + return err + } + + providerResource = res + providerSampler = sdktrace.ParentBased(sdktrace.TraceIDRatioBased(samplingRate())) + + limits := sdktrace.NewSpanLimits() + limits.AttributeCountLimit = utils.GetIntEnvOrDefault("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", 64) + limits.AttributeValueLengthLimit = utils.GetIntEnvOrDefault("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", 1024) + limits.LinkCountLimit = utils.GetIntEnvOrDefault("OTEL_SPAN_LINK_COUNT_LIMIT", 16384) + + opts := []sdktrace.TracerProviderOption{ + sdktrace.WithResource(res), + sdktrace.WithSampler(providerSampler), + sdktrace.WithSpanProcessor(&RHAttributeSpanProcessor{}), + sdktrace.WithRawSpanLimits(limits), + } + + if useBatch { + opts = append(opts, sdktrace.WithBatcher(exp, + sdktrace.WithMaxQueueSize(utils.GetIntEnvOrDefault("OTEL_BSP_MAX_QUEUE_SIZE", 8192)), + sdktrace.WithMaxExportBatchSize(utils.GetIntEnvOrDefault("OTEL_BSP_MAX_EXPORT_BATCH_SIZE", 256)), + sdktrace.WithBatchTimeout(durationFromMsEnv("OTEL_BSP_SCHEDULE_DELAY", 2*time.Second)), + sdktrace.WithExportTimeout(durationFromMsEnv("OTEL_BSP_EXPORT_TIMEOUT", 10*time.Second)), + )) + } else { + opts = append(opts, sdktrace.WithSyncer(exp)) + } + + tracerProvider = sdktrace.NewTracerProvider(opts...) + otel.SetTracerProvider(tracerProvider) + otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator( + propagation.TraceContext{}, + propagation.Baggage{}, + )) + enabled = true + return nil +} + +func Shutdown(ctx context.Context) error { + if tracerProvider == nil { + return nil + } + err := tracerProvider.Shutdown(ctx) + tracerProvider = nil + providerResource = nil + providerSampler = nil + enabled = false + return err +} + +func Tracer() trace.Tracer { + return otel.Tracer("app/base/telemetry") +} + +func newResource() (*resource.Resource, error) { + serviceName := os.Getenv("OTEL_SERVICE_NAME") + if serviceName == "" { + serviceName = RHServiceName + } + version := os.Getenv("IMAGE_TAG") + if version == "" { + version = "unknown" + } + deployEnv := os.Getenv("NAMESPACE") + if deployEnv == "" { + deployEnv = "development" + } + return resource.New(context.Background(), + resource.WithAttributes( + semconv.ServiceName(serviceName), + semconv.ServiceVersion(version), + semconv.DeploymentEnvironment(deployEnv), + ), + ) +} + +func otlpCompressionOptions() []otlptracehttp.Option { + comp := strings.ToLower(os.Getenv("OTEL_EXPORTER_OTLP_COMPRESSION")) + if comp == "" || comp == "none" { + return nil + } + return []otlptracehttp.Option{otlptracehttp.WithCompression(otlptracehttp.GzipCompression)} +} + +func durationFromMsEnv(key string, def time.Duration) time.Duration { + ms := utils.GetIntEnvOrDefault(key, int(def/time.Millisecond)) + return time.Duration(ms) * time.Millisecond +} + +func fmtAttrs(attrs []attribute.KeyValue) []string { + result := make([]string, 0, len(attrs)) + for _, a := range attrs { + result = append(result, fmt.Sprintf("%s=%s", a.Key, a.Value.AsString())) + } + return result +} + +func mustTraceID(s string) trace.TraceID { + id, err := trace.TraceIDFromHex(s) + if err != nil { + panic(err) + } + return id +} + +func mustSpanID(s string) trace.SpanID { + id, err := trace.SpanIDFromHex(s) + if err != nil { + panic(err) + } + return id +} diff --git a/base/telemetry/init_test.go b/base/telemetry/init_test.go new file mode 100644 index 000000000..ab293dd39 --- /dev/null +++ b/base/telemetry/init_test.go @@ -0,0 +1,95 @@ +package telemetry + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" +) + +func TestSQLEnabledAfterInit(t *testing.T) { + t.Setenv("OTEL_ENABLED", "true") + + exp := tracetest.NewInMemoryExporter() + require.NoError(t, initWithExporter(exp)) + defer func() { + _ = Shutdown(context.Background()) + }() + + assert.True(t, SQLEnabled()) +} + +func TestInitDisabledByDefault(t *testing.T) { + t.Setenv("OTEL_ENABLED", "") + require.NoError(t, Init()) + defer func() { + _ = Shutdown(context.Background()) + }() + + assert.False(t, Enabled()) + _, span := Tracer().Start(context.Background(), "noop") + assert.False(t, span.SpanContext().IsValid() && span.IsRecording()) + span.End() +} + +func TestInitPreservesSpanEventLimits(t *testing.T) { + t.Setenv("OTEL_ENABLED", "true") + + exp := tracetest.NewInMemoryExporter() + require.NoError(t, initWithExporter(exp)) + defer func() { + _ = Shutdown(context.Background()) + }() + + _, span := Tracer().Start(context.Background(), "event-limit-test") + span.RecordError(errors.New("test error")) + span.End() + require.NoError(t, tracerProvider.ForceFlush(context.Background())) + + spans := exp.GetSpans() + require.Len(t, spans, 1) + assert.Greater(t, len(spans[0].Events), 0) + assert.Equal(t, 0, spans[0].DroppedEvents) +} + +func TestInitSetsResourceAndSampler(t *testing.T) { + t.Setenv("OTEL_ENABLED", "true") + t.Setenv("OTEL_SERVICE_NAME", "patchman-listener") + t.Setenv("IMAGE_TAG", "test-tag") + t.Setenv("OTEL_SAMPLING_RATE", "0.5") + + exp := tracetest.NewInMemoryExporter() + require.NoError(t, initWithExporter(exp)) + defer func() { + _ = Shutdown(context.Background()) + }() + + res := providerResource + require.NotNil(t, res) + assert.Contains(t, fmtAttrs(res.Attributes()), "service.name=patchman-listener") + assert.Contains(t, fmtAttrs(res.Attributes()), "service.version=test-tag") + + sampler := providerSampler + sampledParent := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: mustTraceID("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + SpanID: mustSpanID("bbbbbbbbbbbbbbbb"), + TraceFlags: trace.FlagsSampled, + Remote: true, + }) + unsampledParent := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: mustTraceID("cccccccccccccccccccccccccccccccc"), + SpanID: mustSpanID("dddddddddddddddd"), + Remote: true, + }) + assert.Equal(t, sdktrace.RecordAndSample, sampler.ShouldSample(sdktrace.SamplingParameters{ + ParentContext: trace.ContextWithRemoteSpanContext(context.Background(), sampledParent), + }).Decision) + assert.Equal(t, sdktrace.Drop, sampler.ShouldSample(sdktrace.SamplingParameters{ + ParentContext: trace.ContextWithRemoteSpanContext(context.Background(), unsampledParent), + }).Decision) +} diff --git a/base/telemetry/kafka.go b/base/telemetry/kafka.go new file mode 100644 index 000000000..7774d44aa --- /dev/null +++ b/base/telemetry/kafka.go @@ -0,0 +1,150 @@ +package telemetry + +import ( + "context" + "strings" + + "github.com/segmentio/kafka-go" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" +) + +type headerCarrier []kafka.Header + +func (c headerCarrier) Get(key string) string { + for _, h := range c { + if strings.EqualFold(h.Key, key) { + return string(h.Value) + } + } + return "" +} + +func (c *headerCarrier) Set(key, value string) { + for i, h := range *c { + if strings.EqualFold(h.Key, key) { + (*c)[i] = kafka.Header{Key: h.Key, Value: []byte(value)} + return + } + } + *c = append(*c, kafka.Header{Key: key, Value: []byte(value)}) +} + +func (c headerCarrier) Keys() []string { + keys := make([]string, len(c)) + for i, h := range c { + keys[i] = h.Key + } + return keys +} + +func Extract(ctx context.Context, headers []kafka.Header) context.Context { + c := headerCarrier(headers) + return otel.GetTextMapPropagator().Extract(ctx, &c) +} + +func Inject(ctx context.Context, headers []kafka.Header) []kafka.Header { + c := headerCarrier(headers) + otel.GetTextMapPropagator().Inject(ctx, &c) + return []kafka.Header(c) +} + +func EncodeTraceparent(ctx context.Context) string { + sc := trace.SpanContextFromContext(ctx) + if !sc.IsValid() { + return "" + } + headers := Inject(ctx, nil) + for _, h := range headers { + if strings.EqualFold(h.Key, "traceparent") { + return string(h.Value) + } + } + return "" +} + +func ContextFromTraceparent(traceparent string) context.Context { + if traceparent == "" { + return context.Background() + } + return Extract(context.Background(), []kafka.Header{ + {Key: "traceparent", Value: []byte(traceparent)}, + }) +} + +func LinksFromTraceparents(tps []string) []trace.Link { + var links []trace.Link + for _, tp := range tps { + if tp == "" { + continue + } + orig := ContextFromTraceparent(tp) + if sc := trace.SpanContextFromContext(orig); sc.IsValid() { + links = append(links, trace.Link{SpanContext: sc}) + } + } + return links +} + +func ConsumerContext(ctx context.Context, topic string, headers []kafka.Header) (context.Context, trace.Span) { + if !Enabled() || !MQEnabled() { + return ctx, trace.SpanFromContext(ctx) + } + ctx = Extract(ctx, headers) + return Tracer().Start(ctx, "process "+topic, + trace.WithSpanKind(trace.SpanKindConsumer), + trace.WithAttributes( + attribute.String("messaging.system", "kafka"), + attribute.String("messaging.destination.name", topic), + attribute.String("messaging.operation", "process"), + ), + ) +} + +func ProducerContext(ctx context.Context, topic string, links []trace.Link) (context.Context, trace.Span) { + if !Enabled() || !MQEnabled() { + return ctx, trace.SpanFromContext(ctx) + } + opts := []trace.SpanStartOption{ + trace.WithSpanKind(trace.SpanKindProducer), + trace.WithAttributes( + attribute.String("messaging.system", "kafka"), + attribute.String("messaging.destination.name", topic), + attribute.String("messaging.operation", "send"), + ), + } + if len(links) > 0 { + opts = append(opts, trace.WithLinks(links...)) + } + return Tracer().Start(ctx, "send "+topic, opts...) +} + +func ItemContext(parent context.Context, name, traceparent string) (context.Context, trace.Span) { + if !Enabled() { + return parent, trace.SpanFromContext(parent) + } + opts := []trace.SpanStartOption{trace.WithSpanKind(trace.SpanKindInternal)} + if orig := ContextFromTraceparent(traceparent); trace.SpanContextFromContext(orig).IsValid() { + opts = append(opts, trace.WithLinks(trace.LinkFromContext(orig))) + } + return Tracer().Start(parent, name, opts...) +} + +func SetRHAttributes(span trace.Span, orgID, requestID string) { + if orgID != "" { + span.SetAttributes(attribute.String("rh.org_id", orgID)) + } + if requestID != "" { + span.SetAttributes(attribute.String("rh.request_id", requestID)) + } +} + +func End(span trace.Span, err error) { + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + } + span.End() +} diff --git a/base/telemetry/kafka_test.go b/base/telemetry/kafka_test.go new file mode 100644 index 000000000..2786da339 --- /dev/null +++ b/base/telemetry/kafka_test.go @@ -0,0 +1,165 @@ +package telemetry + +import ( + "context" + "strings" + "testing" + + "github.com/segmentio/kafka-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" +) + +func setupKafkaTest(t *testing.T) *tracetest.InMemoryExporter { + t.Helper() + t.Setenv("OTEL_ENABLED", "true") + t.Setenv("OTEL_SAMPLING_RATE", "1.0") + exp := tracetest.NewInMemoryExporter() + require.NoError(t, initWithExporter(exp)) + t.Cleanup(func() { _ = Shutdown(context.Background()) }) + return exp +} + +func flushSpans(t *testing.T, exp *tracetest.InMemoryExporter) []tracetest.SpanStub { + t.Helper() + if tracerProvider != nil { + require.NoError(t, tracerProvider.ForceFlush(context.Background())) + } + return exp.GetSpans() +} + +func TestConsumerContextIsChildNotLink(t *testing.T) { + exp := setupKafkaTest(t) + + remote := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: mustTraceID("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + SpanID: mustSpanID("bbbbbbbbbbbbbbbb"), + TraceFlags: trace.FlagsSampled, + Remote: true, + }) + headers := Inject(trace.ContextWithRemoteSpanContext(context.Background(), remote), nil) + + _, span := ConsumerContext(context.Background(), "platform.inventory.events", headers) + span.End() + + spans := flushSpans(t, exp) + require.Len(t, spans, 1) + child := spans[0] + assert.Equal(t, remote.TraceID(), child.SpanContext.TraceID()) + assert.Equal(t, remote.SpanID(), child.Parent.SpanID()) + assert.Empty(t, child.Links) +} + +func TestProducerContextLinksOriginalsAndInjectsOwnTraceparent(t *testing.T) { + exp := setupKafkaTest(t) + + ctx0, item0 := Tracer().Start(context.Background(), "item-0") + tp0 := EncodeTraceparent(ctx0) + item0.End() + + ctx1, item1 := Tracer().Start(context.Background(), "item-1") + tp1 := EncodeTraceparent(ctx1) + item1.End() + + links := LinksFromTraceparents([]string{tp0, tp1}) + producerCtx, producer := ProducerContext(context.Background(), "patchman.evaluator.upload", links) + producerHeaders := Inject(producerCtx, nil) + producer.End() + + spans := flushSpans(t, exp) + var producerStub tracetest.SpanStub + for _, s := range spans { + if s.Name == "send patchman.evaluator.upload" { + producerStub = s + } + } + require.NotEmpty(t, producerStub.Name) + assert.False(t, producerStub.Parent.IsValid()) + require.Len(t, producerStub.Links, 2) + + item0SC := trace.SpanContextFromContext(ContextFromTraceparent(tp0)) + item1SC := trace.SpanContextFromContext(ContextFromTraceparent(tp1)) + linkIDs := []trace.SpanID{producerStub.Links[0].SpanContext.SpanID(), producerStub.Links[1].SpanContext.SpanID()} + assert.Contains(t, linkIDs, item0SC.SpanID()) + assert.Contains(t, linkIDs, item1SC.SpanID()) + + extracted := Extract(context.Background(), producerHeaders) + _, child := Tracer().Start(extracted, "downstream") + child.End() + + spans = flushSpans(t, exp) + var childStub tracetest.SpanStub + for _, s := range spans { + if s.Name == "downstream" { + childStub = s + } + } + require.NotEmpty(t, childStub.Name) + assert.Equal(t, producerStub.SpanContext.TraceID(), childStub.SpanContext.TraceID()) +} + +func TestItemContextLinksOriginalAndParentsBulk(t *testing.T) { + exp := setupKafkaTest(t) + + origCtx, orig := Tracer().Start(context.Background(), "original-item") + originalTraceparent := EncodeTraceparent(origCtx) + orig.End() + + producerCtx, producer := ProducerContext(context.Background(), "patchman.evaluator.upload", nil) + producerHeaders := Inject(producerCtx, nil) + producer.End() + + bulkCtx, bulk := ConsumerContext(context.Background(), "patchman.evaluator.upload", producerHeaders) + _, item := ItemContext(bulkCtx, "evaluate upload", originalTraceparent) + item.End() + bulk.End() + + spans := flushSpans(t, exp) + var bulkStub, itemStub tracetest.SpanStub + for _, s := range spans { + switch s.Name { + case "process patchman.evaluator.upload": + bulkStub = s + case "evaluate upload": + itemStub = s + } + } + require.NotEmpty(t, bulkStub.Name) + require.NotEmpty(t, itemStub.Name) + assert.Equal(t, bulkStub.SpanContext.SpanID(), itemStub.Parent.SpanID()) + + origSC := trace.SpanContextFromContext(ContextFromTraceparent(originalTraceparent)) + require.Len(t, itemStub.Links, 1) + assert.Equal(t, origSC.TraceID(), itemStub.Links[0].SpanContext.TraceID()) + assert.Equal(t, origSC.SpanID(), itemStub.Links[0].SpanContext.SpanID()) +} + +func TestEncodeRoundTrip(t *testing.T) { + setupKafkaTest(t) + + ctx, span := Tracer().Start(context.Background(), "x") + defer span.End() + tp := EncodeTraceparent(ctx) + require.Len(t, strings.Split(tp, "-")[1], 32) + got := trace.SpanContextFromContext(ContextFromTraceparent(tp)) + assert.Equal(t, span.SpanContext().TraceID(), got.TraceID()) + assert.Equal(t, span.SpanContext().SpanID(), got.SpanID()) +} + +func TestHeaderCarrierCaseInsensitive(t *testing.T) { + setupKafkaTest(t) + + headers := []kafka.Header{{Key: "TraceParent", Value: []byte("old")}} + remote := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: mustTraceID("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + SpanID: mustSpanID("bbbbbbbbbbbbbbbb"), + TraceFlags: trace.FlagsSampled, + Remote: true, + }) + out := Inject(trace.ContextWithRemoteSpanContext(context.Background(), remote), headers) + require.Len(t, out, 1) + assert.Equal(t, "TraceParent", out[0].Key) + assert.NotEqual(t, "old", string(out[0].Value)) +} diff --git a/base/telemetry/processor.go b/base/telemetry/processor.go new file mode 100644 index 000000000..741b4049f --- /dev/null +++ b/base/telemetry/processor.go @@ -0,0 +1,43 @@ +package telemetry + +import ( + "context" + + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/trace" +) + +type RHAttributeSpanProcessor struct{} + +func (p *RHAttributeSpanProcessor) OnStart(parent context.Context, s sdktrace.ReadWriteSpan) { + s.SetAttributes(attribute.String("rh.service", RHServiceName)) + + parentSpan := trace.SpanFromContext(parent) + if parentSpan == nil { + return + } + + readOnly, ok := parentSpan.(sdktrace.ReadOnlySpan) + if !ok { + return + } + + currentKeys := make(map[string]bool) + for _, a := range s.Attributes() { + currentKeys[string(a.Key)] = true + } + + for _, a := range readOnly.Attributes() { + key := string(a.Key) + if (key == "rh.org_id" || key == "rh.request_id") && !currentKeys[key] { + s.SetAttributes(a) + } + } +} + +func (p *RHAttributeSpanProcessor) OnEnd(sdktrace.ReadOnlySpan) {} + +func (p *RHAttributeSpanProcessor) Shutdown(context.Context) error { return nil } + +func (p *RHAttributeSpanProcessor) ForceFlush(context.Context) error { return nil } diff --git a/base/telemetry/processor_test.go b/base/telemetry/processor_test.go new file mode 100644 index 000000000..43c9bd725 --- /dev/null +++ b/base/telemetry/processor_test.go @@ -0,0 +1,55 @@ +package telemetry + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" +) + +func TestRHAttributeProcessorSetsServiceAndCopiesParent(t *testing.T) { + exp := tracetest.NewInMemoryExporter() + tp := sdktrace.NewTracerProvider( + sdktrace.WithSyncer(exp), + sdktrace.WithSampler(sdktrace.AlwaysSample()), + sdktrace.WithSpanProcessor(&RHAttributeSpanProcessor{}), + ) + ctx, parent := tp.Tracer("test").Start(context.Background(), "parent") + parent.SetAttributes( + attribute.String("rh.org_id", "org-1"), + attribute.String("rh.request_id", "req-1"), + ) + _, child := tp.Tracer("test").Start(ctx, "child") + child.End() + parent.End() + require.NoError(t, tp.ForceFlush(context.Background())) + + spans := exp.GetSpans() + require.Len(t, spans, 2) + var parentStub, childStub tracetest.SpanStub + for _, s := range spans { + switch s.Name { + case "parent": + parentStub = s + case "child": + childStub = s + } + } + assert.Equal(t, "patchman-engine", attr(parentStub, "rh.service")) + assert.Equal(t, "patchman-engine", attr(childStub, "rh.service")) + assert.Equal(t, "org-1", attr(childStub, "rh.org_id")) + assert.Equal(t, "req-1", attr(childStub, "rh.request_id")) +} + +func attr(s tracetest.SpanStub, key string) string { + for _, a := range s.Attributes { + if string(a.Key) == key { + return a.Value.AsString() + } + } + return "" +} diff --git a/base/utils/gin.go b/base/utils/gin.go index 0cf905596..2dc5014f3 100644 --- a/base/utils/gin.go +++ b/base/utils/gin.go @@ -23,6 +23,10 @@ const ( ReadHeaderTimeout = 60 * time.Second ) +// InstrumentHTTPHandler wraps the server handler (e.g. otelhttp). Default is identity; +// telemetry.Init replaces this with telemetry.InstrumentHandler. +var InstrumentHTTPHandler = func(h http.Handler) http.Handler { return h } + func LoadParamInt(c *gin.Context, param string, defaultValue int, query bool) (int, error) { var valueStr string if query { @@ -103,7 +107,12 @@ type ErrorResponse struct { func RunServer(ctx context.Context, handler http.Handler, port int) error { addr := fmt.Sprintf(":%d", port) - srv := http.Server{Addr: addr, Handler: handler, ReadHeaderTimeout: ReadHeaderTimeout, MaxHeaderBytes: 65535} + srv := http.Server{ + Addr: addr, + Handler: InstrumentHTTPHandler(handler), + ReadHeaderTimeout: ReadHeaderTimeout, + MaxHeaderBytes: 65535, + } go func() { <-ctx.Done() LogDebug("gracefully shutting down server...") diff --git a/base/utils/log.go b/base/utils/log.go index 9851fc79d..68e6dbe3e 100644 --- a/base/utils/log.go +++ b/base/utils/log.go @@ -1,12 +1,14 @@ package utils import ( + "context" "fmt" "os" "time" sentry "github.com/getsentry/sentry-go" log "github.com/sirupsen/logrus" + "go.opentelemetry.io/otel/trace" ) // configure logging using env variables @@ -39,11 +41,23 @@ func processArgs(args []interface{}) (log.Fields, interface{}) { // implement LogXXXX functions to enable additional log fields // usage: utils.LogInfo("my_field_1", 1, "my_field_2", 4.3, "Testing logging") +// Optional leading context.Context adds trace_id / span_id when the span is valid. func logLevel(level log.Level, args ...interface{}) { if !log.IsLevelEnabled(level) { return } + var spanCtx trace.SpanContext + if len(args) > 0 { + if ctx, ok := args[0].(context.Context); ok { + spanCtx = trace.SpanFromContext(ctx).SpanContext() + args = args[1:] + } + } fields, msg := processArgs(args) + if spanCtx.IsValid() { + fields["trace_id"] = spanCtx.TraceID().String() + fields["span_id"] = spanCtx.SpanID().String() + } tryCaptureSentryException(level, fields, msg) diff --git a/base/utils/log_test.go b/base/utils/log_test.go index 6fa10dae2..dd05a199d 100644 --- a/base/utils/log_test.go +++ b/base/utils/log_test.go @@ -1,11 +1,15 @@ package utils import ( + "context" "os" + "regexp" "testing" log "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + sdktrace "go.opentelemetry.io/otel/sdk/trace" ) func TestInitLogging(t *testing.T) { @@ -40,3 +44,30 @@ func TestEvenArgs(t *testing.T) { assert.Equal(t, 1, entry.Data["num"]) assert.Equal(t, "text", entry.Data["str"]) } + +func TestLogInfoWithSpanContext(t *testing.T) { + assert.Nil(t, os.Setenv("LOG_STYLE", "json")) + ConfigureLogging() + + var hook = NewTestLogHook() + log.AddHook(hook) + + tp := sdktrace.NewTracerProvider() + defer func() { _ = tp.Shutdown(context.Background()) }() + ctx, span := tp.Tracer("test").Start(context.Background(), "log-test") + defer span.End() + + LogInfo(ctx, "k", "v", "msg") + + require.Equal(t, 1, len(hook.LogEntries)) + entry := hook.LogEntries[0] + assert.Equal(t, "msg", entry.Message) + assert.Equal(t, "v", entry.Data["k"]) + + traceID, ok := entry.Data["trace_id"].(string) + require.True(t, ok) + spanID, ok := entry.Data["span_id"].(string) + require.True(t, ok) + assert.Regexp(t, regexp.MustCompile(`^[0-9a-f]{32}$`), traceID) + assert.Regexp(t, regexp.MustCompile(`^[0-9a-f]{16}$`), spanID) +} diff --git a/deploy/clowdapp.yaml b/deploy/clowdapp.yaml index 706c3516d..0275dc335 100644 --- a/deploy/clowdapp.yaml +++ b/deploy/clowdapp.yaml @@ -41,6 +41,27 @@ objects: - {name: GOMEMLIMIT, value: '${GOMEMLIMIT_DATABASE_ADMIN}'} - {name: POD_CONFIG, value: '${ADMIN_CONFIG}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-admin'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} resources: limits: {cpu: '${CPU_LIMIT_ADMIN}', memory: '${MEM_LIMIT_ADMIN}'} requests: {cpu: '${CPU_REQUEST_ADMIN}', memory: '${MEM_REQUEST_ADMIN}'} @@ -105,6 +126,27 @@ objects: - {name: KESSEL_AUTH_CLIENT_ID, valueFrom: {secretKeyRef: {name: kessel-service-client, key: id}}} - {name: KESSEL_AUTH_CLIENT_SECRET, valueFrom: {secretKeyRef: {name: kessel-service-client, key: secret}}} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-manager'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} resources: limits: {cpu: '${CPU_LIMIT_MANAGER}', memory: '${MEM_LIMIT_MANAGER}'} requests: {cpu: '${CPU_REQUEST_MANAGER}', memory: '${MEM_REQUEST_MANAGER}'} @@ -156,6 +198,27 @@ objects: - {name: POD_CONFIG, value: '${LISTENER_CONFIG}'} - {name: CONTENT_SOURCES_USER, value: '${CONTENT_SOURCES_USER}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-listener'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} resources: limits: {cpu: '${CPU_LIMIT_LISTENER}', memory: '${MEM_LIMIT_LISTENER}'} requests: {cpu: '${CPU_REQUEST_LISTENER}', memory: '${MEM_REQUEST_LISTENER}'} @@ -205,6 +268,27 @@ objects: - {name: GOMEMLIMIT, value: '${GOMEMLIMIT_EVALUATOR}'} - {name: POD_CONFIG, value: 'label=upload;${EVALUATOR_UPLOAD_CONFIG}'} - {name: CONSOLEDOT_HOSTNAME, value: '${CONSOLEDOT_HOSTNAME}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-evaluator-upload'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} resources: limits: {cpu: '${CPU_LIMIT_EVALUATOR_UPLOAD}', memory: '${MEM_LIMIT_EVALUATOR_UPLOAD}'} requests: {cpu: '${CPU_REQUEST_EVALUATOR_UPLOAD}', memory: '${MEM_REQUEST_EVALUATOR_UPLOAD}'} @@ -254,6 +338,27 @@ objects: - {name: GOMEMLIMIT, value: '${GOMEMLIMIT_EVALUATOR}'} - {name: POD_CONFIG, value: 'label=recalc;payload_tracker=false;${EVALUATOR_RECALC_CONFIG}'} - {name: CONSOLEDOT_HOSTNAME, value: '${CONSOLEDOT_HOSTNAME}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-evaluator-recalc'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} resources: limits: {cpu: '${CPU_LIMIT_EVALUATOR_RECALC}', memory: '${MEM_LIMIT_EVALUATOR_RECALC}'} requests: {cpu: '${CPU_REQUEST_EVALUATOR_RECALC}', memory: '${MEM_REQUEST_EVALUATOR_RECALC}'} @@ -303,6 +408,27 @@ objects: - {name: GOMEMLIMIT, value: '${GOMEMLIMIT_EVALUATOR}'} - {name: POD_CONFIG, value: 'label=user-evaluation;payload_tracker=false;${EVALUATOR_USER_EVALUATION_CONFIG}'} - {name: CONSOLEDOT_HOSTNAME, value: '${CONSOLEDOT_HOSTNAME}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-evaluator-user-evaluation'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} resources: limits: {cpu: '${CPU_LIMIT_EVALUATOR_USER_EVALUATION}', memory: '${MEM_LIMIT_EVALUATOR_USER_EVALUATION}'} requests: {cpu: '${CPU_REQUEST_EVALUATOR_USER_EVALUATION}', memory: '${MEM_REQUEST_EVALUATOR_USER_EVALUATION}'} @@ -348,6 +474,27 @@ objects: - {name: GOMEMLIMIT, value: '${GOMEMLIMIT_AGGREGATOR}'} - {name: POD_CONFIG, value: 'label=aggregator;${AGGREGATOR_CONFIG}'} - {name: CONSOLEDOT_HOSTNAME, value: '${CONSOLEDOT_HOSTNAME}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-aggregator'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} resources: limits: {cpu: '${CPU_LIMIT_AGGREGATOR}', memory: '${MEM_LIMIT_AGGREGATOR}'} requests: {cpu: '${CPU_REQUEST_AGGREGATOR}', memory: '${MEM_REQUEST_AGGREGATOR}'} @@ -377,6 +524,27 @@ objects: - {name: VMAAS_SYNC_PASSWORD, valueFrom: {secretKeyRef: {name: patchman-engine-database-passwords, key: vmaas-sync-database-password}}} - {name: POD_CONFIG, value: '${DATABASE_ADMIN_CONFIG}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-db-migration'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} resources: limits: {cpu: '${CPU_LIMIT_DATABASE_ADMIN}', memory: '${MEM_LIMIT_DATABASE_ADMIN}'} requests: {cpu: '${CPU_REQUEST_DATABASE_ADMIN}', memory: '${MEM_REQUEST_DATABASE_ADMIN}'} @@ -417,6 +585,27 @@ objects: - {name: POD_CONFIG, value: '${JOBS_CONFIG}'} - {name: CANDLEPIN_CERT, valueFrom: {secretKeyRef: {name: candlepin, key: cert}}} - {name: CANDLEPIN_KEY, valueFrom: {secretKeyRef: {name: candlepin, key: key}}} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-vmaas-sync'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} resources: limits: {cpu: '${CPU_LIMIT_VMAAS_SYNC}', memory: '${MEM_LIMIT_VMAAS_SYNC}'} requests: {cpu: '${CPU_REQUEST_VMAAS_SYNC}', memory: '${MEM_REQUEST_VMAAS_SYNC}'} @@ -449,6 +638,27 @@ objects: key: vmaas-sync-database-password}}} - {name: PROMETHEUS_PUSHGATEWAY,value: '${PROMETHEUS_PUSHGATEWAY}'} - {name: POD_CONFIG, value: '${JOBS_CONFIG}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-system-culling'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} - name: package-refresh activeDeadlineSeconds: ${{JOBS_TIMEOUT}} @@ -484,6 +694,27 @@ objects: - {name: DB_WORK_MEM, value: '${DB_WORK_MEM}'} - {name: PROMETHEUS_PUSHGATEWAY,value: '${PROMETHEUS_PUSHGATEWAY}'} - {name: POD_CONFIG, value: '${JOBS_CONFIG}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-package-refresh'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} - name: advisory-refresh activeDeadlineSeconds: ${{JOBS_TIMEOUT}} @@ -512,6 +743,27 @@ objects: - {name: DB_PASSWD, valueFrom: {secretKeyRef: {name: patchman-engine-database-passwords, key: vmaas-sync-database-password}}} - {name: POD_CONFIG, value: '${JOBS_CONFIG}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-advisory-refresh'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} - name: delete-unused activeDeadlineSeconds: ${{JOBS_TIMEOUT}} @@ -540,6 +792,27 @@ objects: - {name: DB_PASSWD, valueFrom: {secretKeyRef: {name: patchman-engine-database-passwords, key: vmaas-sync-database-password}}} - {name: POD_CONFIG, value: '${JOBS_CONFIG}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-delete-unused'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} - name: repack activeDeadlineSeconds: ${{JOBS_TIMEOUT}} @@ -565,6 +838,27 @@ objects: - {name: SENTRY_DSN, valueFrom: {secretKeyRef: {name: patchman-sentry, key: sentry-dsn}}} - {name: DB_DEBUG, value: '${DB_DEBUG_JOBS}'} - {name: POD_CONFIG, value: '${JOBS_CONFIG}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-repack'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} - name: clean-advisory-account-data activeDeadlineSeconds: ${{JOBS_TIMEOUT}} @@ -593,6 +887,27 @@ objects: - {name: DB_PASSWD, valueFrom: {secretKeyRef: {name: patchman-engine-database-passwords, key: vmaas-sync-database-password}}} - {name: POD_CONFIG, value: '${JOBS_CONFIG}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-clean-advisory-account-data'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} - name: account-advisory-backfill activeDeadlineSeconds: ${{JOBS_TIMEOUT}} @@ -621,6 +936,27 @@ objects: - {name: DB_PASSWD, valueFrom: {secretKeyRef: {name: patchman-engine-database-passwords, key: vmaas-sync-database-password}}} - {name: POD_CONFIG, value: '${JOBS_CONFIG}'} + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-account-advisory-backfill'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} - name: system-advisories-0-recovery # One-shot Job (no schedule): runs on deploy / CJI like db-migration. @@ -655,6 +991,27 @@ objects: - {name: SSL_CERT_DIR, value: '${SSL_CERT_DIR}'} - {name: POD_CONFIG, value: '${JOBS_CONFIG}'} # set system_advisories_0_recovery=true for cutover + - {name: IMAGE_TAG, value: '${IMAGE_TAG}'} + - {name: OTEL_SERVICE_NAME, value: 'patchman-system-advisories-0-recovery'} + - {name: OTEL_ENABLED, value: '${OTEL_ENABLED}'} + - {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: '${OTEL_EXPORTER_OTLP_ENDPOINT}'} + - {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: '${OTEL_EXPORTER_OTLP_PROTOCOL}'} + - {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: '${OTEL_EXPORTER_OTLP_COMPRESSION}'} + - {name: OTEL_TRACES_EXPORTER, value: '${OTEL_TRACES_EXPORTER}'} + - {name: OTEL_METRICS_EXPORTER, value: '${OTEL_METRICS_EXPORTER}'} + - {name: OTEL_LOGS_EXPORTER, value: '${OTEL_LOGS_EXPORTER}'} + - {name: OTEL_SAMPLING_RATE, value: '${OTEL_SAMPLING_RATE}'} + - {name: OTEL_SQL_ENABLED, value: '${OTEL_SQL_ENABLED}'} + - {name: OTEL_HTTP_INBOUND_ENABLED, value: '${OTEL_HTTP_INBOUND_ENABLED}'} + - {name: OTEL_HTTP_OUTBOUND_ENABLED, value: '${OTEL_HTTP_OUTBOUND_ENABLED}'} + - {name: OTEL_MQ_ENABLED, value: '${OTEL_MQ_ENABLED}'} + - {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '${OTEL_BSP_MAX_QUEUE_SIZE}'} + - {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '${OTEL_BSP_MAX_EXPORT_BATCH_SIZE}'} + - {name: OTEL_BSP_SCHEDULE_DELAY, value: '${OTEL_BSP_SCHEDULE_DELAY}'} + - {name: OTEL_BSP_EXPORT_TIMEOUT, value: '${OTEL_BSP_EXPORT_TIMEOUT}'} + - {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT}'} + - {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT}'} + - {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '${OTEL_SPAN_LINK_COUNT_LIMIT}'} database: name: patchman version: 16 @@ -983,6 +1340,25 @@ parameters: # Common parameters - {name: IMAGE, value: quay.io/redhat-services-prod/insights-management-tenant/insights-patch/patchman-engine} - {name: IMAGE_TAG, required: true} +- {name: OTEL_ENABLED, value: 'false'} +- {name: OTEL_EXPORTER_OTLP_ENDPOINT, value: ''} +- {name: OTEL_EXPORTER_OTLP_PROTOCOL, value: 'http/protobuf'} +- {name: OTEL_EXPORTER_OTLP_COMPRESSION, value: 'gzip'} +- {name: OTEL_TRACES_EXPORTER, value: 'otlp'} +- {name: OTEL_METRICS_EXPORTER, value: 'none'} +- {name: OTEL_LOGS_EXPORTER, value: 'none'} +- {name: OTEL_SAMPLING_RATE, value: '1.0'} +- {name: OTEL_SQL_ENABLED, value: 'true'} +- {name: OTEL_HTTP_INBOUND_ENABLED, value: 'true'} +- {name: OTEL_HTTP_OUTBOUND_ENABLED, value: 'true'} +- {name: OTEL_MQ_ENABLED, value: 'true'} +- {name: OTEL_BSP_MAX_QUEUE_SIZE, value: '8192'} +- {name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE, value: '256'} +- {name: OTEL_BSP_SCHEDULE_DELAY, value: '2000'} +- {name: OTEL_BSP_EXPORT_TIMEOUT, value: '10000'} +- {name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, value: '64'} +- {name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, value: '1024'} +- {name: OTEL_SPAN_LINK_COUNT_LIMIT, value: '16384'} - {name: ENV_NAME, required: false} - {name: GIN_MODE, value: 'release'} # Gin webframework running mode - {name: KAFKA_READER_MAX_ATTEMPTS, value: '3'} # Limit of how many attempts will be made before kafka read error. diff --git a/evaluator/advisory_update_test.go b/evaluator/advisory_update_test.go index 270cf6966..e26d60228 100644 --- a/evaluator/advisory_update_test.go +++ b/evaluator/advisory_update_test.go @@ -6,6 +6,7 @@ import ( "app/base/models" "app/base/mqueue" "app/base/utils" + "context" "testing" "github.com/bytedance/sonic" @@ -137,7 +138,7 @@ func TestAdvisoryUpdateKafkaRoundTrip(t *testing.T) { }() var received mqueue.KafkaMessage - go reader.HandleMessages(t.Context(), func(m mqueue.KafkaMessage) error { + go reader.HandleMessages(t.Context(), func(_ context.Context, m mqueue.KafkaMessage) error { received = m return nil }) @@ -161,7 +162,7 @@ func TestAdvisoryUpdateKafkaRoundTrip(t *testing.T) { OrgID: &orgID, AccountID: rhAccountID}) assert.NoError(t, err) - err = evaluateHandler(mqueue.KafkaMessage{Value: data}) + err = evaluateHandler(context.Background(), mqueue.KafkaMessage{Value: data}) assert.NoError(t, err) utils.AssertEqualWait(t, 10, func() (exp, act interface{}) { diff --git a/evaluator/evaluate.go b/evaluator/evaluate.go index da6daa90e..69969bb98 100644 --- a/evaluator/evaluate.go +++ b/evaluator/evaluate.go @@ -7,6 +7,7 @@ import ( "app/base/database" "app/base/models" "app/base/mqueue" + "app/base/telemetry" "app/base/types" "app/base/utils" "app/base/vmaas" @@ -19,6 +20,7 @@ import ( "github.com/google/uuid" "github.com/jinzhu/copier" "github.com/pkg/errors" + "go.opentelemetry.io/otel/trace" "gorm.io/gorm" "gorm.io/gorm/clause" @@ -236,7 +238,7 @@ func evaluateInDatabase(ctx context.Context, event *mqueue.PlatformEvent, invent return nil, nil, nil } - vmaasData, err := evaluateWithVmaas(updatesData, system, event) + vmaasData, err := evaluateWithVmaas(ctx, updatesData, system, event) if err != nil { return nil, nil, errors.Wrap(err, "evaluation with vmaas failed") } @@ -287,11 +289,11 @@ func tryGetYumUpdates(system *models.SystemPlatformV2) (*vmaas.UpdatesV3Response return &resp, nil } -func evaluateWithVmaas(updatesData *vmaas.UpdatesV3Response, +func evaluateWithVmaas(ctx context.Context, updatesData *vmaas.UpdatesV3Response, system *models.SystemPlatformV2, event *mqueue.PlatformEvent) (*vmaas.UpdatesV3Response, error) { defer utils.ObserveSecondsSince(time.Now(), evaluationPartDuration.WithLabelValues("evaluate-with-vmaas-full")) - err := evaluateAndStore(system, updatesData, event) + err := evaluateAndStore(ctx, system, updatesData, event) if err != nil { return nil, errors.Wrap(err, "Unable to evaluate and store results") } @@ -477,7 +479,7 @@ func commitWithObserve(tx *gorm.DB) error { // and then executes all deletions, updates, and insertions in a single transaction. //nolint:funlen -func evaluateAndStore(system *models.SystemPlatformV2, +func evaluateAndStore(ctx context.Context, system *models.SystemPlatformV2, vmaasData *vmaas.UpdatesV3Response, event *mqueue.PlatformEvent) error { advisoriesByName, err := lazySaveAndLoadAdvisories(system, vmaasData) if err != nil { @@ -489,7 +491,7 @@ func evaluateAndStore(system *models.SystemPlatformV2, return errors.Wrap(err, "Package loading failed") } - tx := database.DB.WithContext(base.Context).Begin() + tx := database.DB.WithContext(ctx).Begin() // Don't allow requested TX to hang around locking the rows defer tx.Rollback() @@ -742,41 +744,58 @@ func invalidateCaches(orgID string) error { return err } -func evaluateHandler(m mqueue.KafkaMessage) error { +func systemIDsAndTraceparents(event mqueue.PlatformEvent) ([]uuid.UUID, []string) { + if event.SystemIDs != nil { + return event.SystemIDs, event.Traceparents + } + return []uuid.UUID{event.ID}, event.Traceparents +} + +func firstRequestID(event mqueue.PlatformEvent) string { + if len(event.RequestIDs) > 0 { + return event.RequestIDs[0] + } + return "" +} + +func evaluateHandler(ctx context.Context, m mqueue.KafkaMessage) error { var event mqueue.PlatformEvent if err := sonic.Unmarshal(m.Value, &event); err != nil { utils.LogError("err", err, "Could not deserialize platform event") return nil } + span := trace.SpanFromContext(ctx) + telemetry.SetRHAttributes(span, event.GetOrgID(), firstRequestID(event)) + var err error var wg sync.WaitGroup guard := make(chan struct{}, nEvalGoroutines) - nSystems := 1 - if event.SystemIDs != nil { - nSystems = len(event.SystemIDs) - } - ptEvents := make(mqueue.PayloadTrackerEvents, 0, nSystems) + ids, tps := systemIDsAndTraceparents(event) + ptEvents := make(mqueue.PayloadTrackerEvents, 0, len(ids)) ptEvent := mqueue.PayloadTrackerEvent{ OrgID: event.OrgID, Status: "success", StatusMsg: "advisories evaluation", } - if event.SystemIDs != nil { - // Evaluate in bulk - nRequestIDs := len(event.RequestIDs) - for i, id := range event.SystemIDs { - ptEvent.InventoryID = id - if nRequestIDs > i { - ptEvent.RequestID = &event.RequestIDs[i] - } - ptEvent, err = runEvaluate(base.Context, event, id, evalLabel, ptEvent, &wg, guard) - ptEvents = append(ptEvents, ptEvent) + nRequestIDs := len(event.RequestIDs) + for i, id := range ids { + ptEvent.InventoryID = id + if nRequestIDs > i { + ptEvent.RequestID = &event.RequestIDs[i] + } + tp := "" + if i < len(tps) { + tp = tps[i] + } + itemCtx, itemSpan := telemetry.ItemContext(ctx, "evaluate "+evalLabel, tp) + if i < len(event.RequestIDs) { + telemetry.SetRHAttributes(itemSpan, event.GetOrgID(), event.RequestIDs[i]) } - } else { - ptEvent, err = runEvaluate(base.Context, event, event.ID, evalLabel, ptEvent, &wg, guard) + ptEvent, err = runEvaluate(itemCtx, event, id, evalLabel, ptEvent, &wg, guard) + telemetry.End(itemSpan, err) ptEvents = append(ptEvents, ptEvent) } wg.Wait() @@ -787,7 +806,7 @@ func evaluateHandler(m mqueue.KafkaMessage) error { // send kafka message to payload tracker if evalLabel == uploadLabel { - ptErr := mqueue.SendMessages(base.Context, ptWriter, &ptEvents) + ptErr := mqueue.SendMessages(ctx, ptWriter, &ptEvents) if ptErr != nil { // don't fail with err, just log that we couldn't send msg to payload tracker utils.LogWarn("err", ptErr, WarnPayloadTracker) diff --git a/evaluator/evaluate_test.go b/evaluator/evaluate_test.go index acb5b819e..062b9da87 100644 --- a/evaluator/evaluate_test.go +++ b/evaluator/evaluate_test.go @@ -27,6 +27,29 @@ func TestInit(_ *testing.T) { utils.TestLoadEnv("conf/evaluator_common.env", "conf/evaluator_upload.env") } +func TestSystemIDsAndTraceparents(t *testing.T) { + id1 := uuid.MustParse("00000000-0000-0000-0000-000000000001") + id2 := uuid.MustParse("00000000-0000-0000-0000-000000000002") + single := uuid.MustParse("00000000-0000-0000-0000-000000000099") + + ids, tps := systemIDsAndTraceparents(mqueue.PlatformEvent{ + SystemIDs: []uuid.UUID{id1, id2}, + Traceparents: []string{"tp1", "tp2"}, + }) + assert.Equal(t, []uuid.UUID{id1, id2}, ids) + assert.Equal(t, []string{"tp1", "tp2"}, tps) + + ids, tps = systemIDsAndTraceparents(mqueue.PlatformEvent{ + ID: single, + Traceparents: []string{"tp-single"}, + }) + assert.Equal(t, []uuid.UUID{single}, ids) + assert.Equal(t, []string{"tp-single"}, tps) + + assert.Equal(t, "r1", firstRequestID(mqueue.PlatformEvent{RequestIDs: []string{"r1", "r2"}})) + assert.Equal(t, "", firstRequestID(mqueue.PlatformEvent{})) +} + // nolint: funlen func TestEvaluate(t *testing.T) { utils.SkipWithoutDB(t) @@ -68,7 +91,7 @@ func TestEvaluate(t *testing.T) { OrgID: &orgID, AccountID: rhAccountID}) assert.NoError(t, err) - err = evaluateHandler(mqueue.KafkaMessage{Value: data}) + err = evaluateHandler(context.Background(), mqueue.KafkaMessage{Value: data}) assert.NoError(t, err) advisoryIDs := database.CheckAdvisoriesInDB(t, expectedAddedAdvisories) @@ -88,7 +111,7 @@ func TestEvaluate(t *testing.T) { OrgID: &orgID, AccountID: rhAccountID}) assert.NoError(t, err) - err = evaluateHandler(mqueue.KafkaMessage{Value: data}) + err = evaluateHandler(context.Background(), mqueue.KafkaMessage{Value: data}) assert.NoError(t, err) database.CheckSystemJustEvaluated(t, testInventoryID, 3, 1, 1, 0, 3, 1, 1, 0, 2, 2, 2, true) @@ -133,7 +156,7 @@ func TestEvaluateYum(t *testing.T) { OrgID: &orgID, AccountID: rhAccountID}) assert.NoError(t, err) - err = evaluateHandler(mqueue.KafkaMessage{Value: data}) + err = evaluateHandler(context.Background(), mqueue.KafkaMessage{Value: data}) assert.NoError(t, err) expectedPackageIDs := database.GetPackageIDs(expectedPackages...) diff --git a/evaluator/notifications_test.go b/evaluator/notifications_test.go index fce75c4a3..8f852b4be 100644 --- a/evaluator/notifications_test.go +++ b/evaluator/notifications_test.go @@ -7,6 +7,7 @@ import ( "app/base/mqueue" ntf "app/base/notification" "app/base/utils" + "context" "fmt" "testing" "time" @@ -64,7 +65,7 @@ func TestAdvisoriesNotificationPublish(t *testing.T) { AccountID: rhAccountID, OrgID: &orgID}) assert.NoError(t, err) - err = evaluateHandler(mqueue.KafkaMessage{Value: data}) + err = evaluateHandler(context.Background(), mqueue.KafkaMessage{Value: data}) assert.NoError(t, err) advisoryIDs := database.CheckAdvisoriesInDB(t, expectedAddedAdvisories) database.CheckAdvisoriesAccountDataNotified(t, rhAccountID, expectedAdvisoryIDs, true) @@ -118,7 +119,7 @@ func TestAdvisoriesNotificationSkipPublishViaEvaluate(t *testing.T) { SkipNotifications: true, }) assert.NoError(t, err) - err = evaluateHandler(mqueue.KafkaMessage{Value: data}) + err = evaluateHandler(context.Background(), mqueue.KafkaMessage{Value: data}) assert.NoError(t, err) advisoryIDs := database.CheckAdvisoriesInDB(t, expectedAddedAdvisories) diff --git a/evaluator/template_advisory_e2e_test.go b/evaluator/template_advisory_e2e_test.go index 6aaf9250e..e019b4ab2 100644 --- a/evaluator/template_advisory_e2e_test.go +++ b/evaluator/template_advisory_e2e_test.go @@ -7,6 +7,7 @@ import ( "app/base/mqueue" "app/base/utils" "app/listener" + "context" "strings" "testing" "time" @@ -30,7 +31,7 @@ func evaluateHandlerForTest(event mqueue.PlatformEvent) error { if err != nil { return err } - return evaluateHandler(mqueue.KafkaMessage{Value: data}) + return evaluateHandler(context.Background(), mqueue.KafkaMessage{Value: data}) } // nolint: funlen @@ -86,7 +87,7 @@ func TestTemplateAdvisoryEvalE2E(t *testing.T) { msg, err := sonic.Marshal(updateEvent) require.NoError(t, err) - err = listener.TemplatesMessageHandler(mqueue.KafkaMessage{Value: msg}) + err = listener.TemplatesMessageHandler(context.Background(), mqueue.KafkaMessage{Value: msg}) require.NoError(t, err) database.CheckTemplateAdvisories(t, template.ID, []int64{1, 3}) diff --git a/go.mod b/go.mod index 3e3d68c94..efd6dc696 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.25.7 require ( github.com/MichaelMraka/gorpm v0.0.0-20251128174203-65cf25f01bac + github.com/XSAM/otelsql v0.41.0 github.com/aws/aws-sdk-go v1.55.8 github.com/bytedance/sonic v1.15.2 github.com/getkin/kin-openapi v0.146.0 @@ -31,6 +32,11 @@ require ( github.com/swaggo/files v1.0.1 github.com/swaggo/gin-swagger v1.6.1 github.com/zsais/go-gin-prometheus v1.0.3 + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.66.0 + go.opentelemetry.io/otel v1.44.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0 + go.opentelemetry.io/otel/sdk v1.44.0 + go.opentelemetry.io/otel/trace v1.44.0 go.uber.org/automaxprocs v1.6.0 go.uber.org/ratelimit v0.3.1 golang.org/x/exp v0.0.0-20260813180055-c1d0aacb2297 @@ -49,10 +55,12 @@ require ( github.com/beorn7/perks v1.0.1 // indirect github.com/bytedance/gopkg v0.1.4 // indirect github.com/bytedance/sonic/loader v0.5.1 // indirect + github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/ezamriy/gorpm v0.0.0-20160905202458-25f7273cbf51 // indirect + github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/gabriel-vasile/mimetype v1.4.13 // indirect github.com/gin-contrib/gzip v1.2.5 // indirect @@ -77,6 +85,7 @@ require ( github.com/goccy/go-json v0.10.6 // indirect github.com/goccy/go-yaml v1.19.2 // indirect github.com/gorilla/securecookie v1.1.2 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/pgx/v5 v5.10.0 // indirect @@ -117,9 +126,9 @@ require ( github.com/zitadel/schema v1.3.2 // indirect go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect - go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect - go.opentelemetry.io/otel/trace v1.44.0 // indirect + go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.uber.org/atomic v1.11.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/arch v0.25.0 // indirect diff --git a/go.sum b/go.sum index 13a8e33b4..08635b95b 100644 --- a/go.sum +++ b/go.sum @@ -10,6 +10,8 @@ github.com/MichaelMraka/gorpm v0.0.0-20251128174203-65cf25f01bac h1:qxH8XNTYCJve github.com/MichaelMraka/gorpm v0.0.0-20251128174203-65cf25f01bac/go.mod h1:R3iHBCKh4OIuNrIzJl+81N08s7dD92MLffdRy+wbtnk= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/XSAM/otelsql v0.41.0 h1:uZifjQhZhv5EDYJh+IVk1DiYxQZJBlNSen0MBFnfxB8= +github.com/XSAM/otelsql v0.41.0/go.mod h1:NMQT0PiKoFILp9QgjQz+D5mvW+9mT0suR7OejqrtMaM= github.com/aws/aws-sdk-go v1.49.13/go.mod h1:LF8svs817+Nz+DmiMQKTO3ubZ/6IaTpq3TjupRn3Eqk= github.com/aws/aws-sdk-go v1.55.8 h1:JRmEUbU52aJQZ2AjX4q4Wu7t4uZjOu71uyNmaWlUkJQ= github.com/aws/aws-sdk-go v1.55.8/go.mod h1:ZkViS9AqA6otK+JBBNH2++sx1sgxrPKcSzPPvQkUtXk= @@ -25,6 +27,8 @@ github.com/bytedance/sonic v1.15.2 h1:90H+rcF/FwLXwfB1cudOLq/je83n683Utf4Cbp0xHC github.com/bytedance/sonic v1.15.2/go.mod h1:mT2NbXunuaEbnZ+mRIX/vYqKISmgEuHFDI4UzmKx2SA= github.com/bytedance/sonic/loader v0.5.1 h1:Ygpfa9zwRCCKSlrp5bBP/b/Xzc3VxsAW+5NIYXrOOpI= github.com/bytedance/sonic/loader v0.5.1/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo= +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M= @@ -158,6 +162,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/securecookie v1.1.2 h1:YCIWL56dvtr73r6715mJs5ZvhtnY73hBvEF8kXD8ePA= github.com/gorilla/securecookie v1.1.2/go.mod h1:NfCASbcHqRSY+3a8tlWJwsQap2VX5pwzwo4h3eOamfo= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= @@ -338,10 +344,14 @@ go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.66.0 h1:PnV4kVnw0zOmwwFkAzCN5O07fw1YOIQor120zrh0AVo= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.66.0/go.mod h1:ofAwF4uinaf8SXdVzzbL4OsxJ3VfeEg3f/F6CeF49/Y= go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 h1:4YsVu3B8+3qtWYYrsUYgn0OG78pN0rnNPRGX4SbokQI= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0/go.mod h1:+wnlSn0mD1ADVMe3v9Z/WIaiz6q6gL2J/ejaAmdmv80= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0 h1:lgh3PiVrRUWMLOVSkQicxzZll5NjF1r+AtsX1XRIHw0= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0/go.mod h1:5Cnhth3m/AgOeTgE3ex12pPmiu/gGtZit03kSzx9X7s= go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= @@ -350,6 +360,8 @@ go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRk go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= +go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= +go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= diff --git a/listener/event_buffers.go b/listener/event_buffers.go index 95216d85e..e4acf26fa 100644 --- a/listener/event_buffers.go +++ b/listener/event_buffers.go @@ -3,7 +3,9 @@ package listener import ( "app/base" "app/base/mqueue" + "app/base/telemetry" "app/base/utils" + "context" "sync" "time" @@ -36,6 +38,7 @@ func (b *eventBuffer) bufferEvalEvents( inventoryID uuid.UUID, rhAccountID int, ptEvent *mqueue.PayloadTrackerEvent, + ctx context.Context, ) { defer utils.ObserveSecondsSince(time.Now(), messagePartDuration.WithLabelValues("buffer-eval-events")) @@ -45,6 +48,7 @@ func (b *eventBuffer) bufferEvalEvents( RhAccountID: rhAccountID, OrgID: ptEvent.OrgID, RequestID: *ptEvent.RequestID, + Traceparent: telemetry.EncodeTraceparent(ctx), } b.evalBuffer = append(b.evalBuffer, evalData) b.ptBuffer = append(b.ptBuffer, *ptEvent) @@ -63,14 +67,22 @@ func (b *eventBuffer) flushEvalEvents() { tStart := time.Now() b.lock.Lock() defer b.lock.Unlock() - err := mqueue.SendMessages(base.Context, *b.evalWriter, b.evalBuffer) + + tps := make([]string, 0, len(b.evalBuffer)) + for _, e := range b.evalBuffer { + tps = append(tps, e.Traceparent) + } + links := telemetry.LinksFromTraceparents(tps) + pctx, span := telemetry.ProducerContext(context.Background(), utils.CoreCfg.EvalTopic, links) + var err error + defer func() { telemetry.End(span, err) }() + err = mqueue.SendMessages(pctx, *b.evalWriter, b.evalBuffer) if err != nil { utils.LogError("err", err, ErrorKafkaSend) } utils.ObserveSecondsSince(tStart, messagePartDuration.WithLabelValues("buffer-sent-evaluator")) - err = mqueue.SendMessages(base.Context, *b.ptWriter, b.ptBuffer) - if err != nil { - utils.LogWarn("err", err, WarnPayloadTracker) + if ptErr := mqueue.SendMessages(base.Context, *b.ptWriter, b.ptBuffer); ptErr != nil { + utils.LogWarn("err", ptErr, WarnPayloadTracker) } utils.ObserveSecondsSince(tStart, messagePartDuration.WithLabelValues("buffer-sent-payload-tracker")) utils.LogDebug("evaluator_messages", len(b.evalBuffer), diff --git a/listener/events.go b/listener/events.go index 6d81147e2..acb0934b8 100644 --- a/listener/events.go +++ b/listener/events.go @@ -4,11 +4,14 @@ import ( "app/base/database" "app/base/models" "app/base/mqueue" + "app/base/telemetry" "app/base/utils" + "context" "time" "github.com/bytedance/sonic" "github.com/pkg/errors" + "go.opentelemetry.io/otel/trace" "gorm.io/gorm" ) @@ -18,7 +21,7 @@ const ( WarnNoRowsModified = "no rows modified on delete event" ) -func EventsMessageHandler(m mqueue.KafkaMessage) error { +func EventsMessageHandler(ctx context.Context, m mqueue.KafkaMessage) error { var msgData map[string]interface{} utils.LogTrace("kafka message data", string(m.Value)) if err := sonic.Unmarshal(m.Value, &msgData); err != nil { @@ -45,7 +48,8 @@ func EventsMessageHandler(m mqueue.KafkaMessage) error { utils.LogError("inventoryID", msgData["id"], "msg", string(m.Value), "Invalid 'delete' message format") } - return HandleDelete(event) + telemetry.SetRHAttributes(trace.SpanFromContext(ctx), event.GetOrgID(), "") + return HandleDelete(ctx, event) case "updated": fallthrough case "created": @@ -55,18 +59,19 @@ func EventsMessageHandler(m mqueue.KafkaMessage) error { "Invalid 'updated' message format") return nil } - return HandleUpload(event) + telemetry.SetRHAttributes(trace.SpanFromContext(ctx), event.Host.GetOrgID(), event.Metadata.RequestID) + return HandleUpload(ctx, event) default: utils.LogWarn("msg", string(m.Value), WarnUnknownType) return nil } } -func HandleDelete(event mqueue.PlatformEvent) error { +func HandleDelete(ctx context.Context, event mqueue.PlatformEvent) error { defer utils.ObserveSecondsSince(time.Now(), messageHandlingDuration.WithLabelValues(EventDelete)) // TODO: Do we need locking here ? - err := database.OnConflictUpdate(database.DB, "inventory_id", "when_deleted"). + err := database.OnConflictUpdate(database.DB.WithContext(ctx), "inventory_id", "when_deleted"). Create(models.DeletedSystem{ InventoryID: event.ID, WhenDeleted: time.Now(), @@ -79,7 +84,7 @@ func HandleDelete(event mqueue.PlatformEvent) error { } // mark system as stale and let system_culling job remove it later - query := database.DB.Model(&models.SystemInventory{}).Where("inventory_id = ?", event.ID). + query := database.DB.WithContext(ctx).Model(&models.SystemInventory{}).Where("inventory_id = ?", event.ID). Updates(map[string]interface{}{ "culled_timestamp": gorm.Expr("NOW()"), }) diff --git a/listener/events_test.go b/listener/events_test.go index 0546a8729..6886f4319 100644 --- a/listener/events_test.go +++ b/listener/events_test.go @@ -6,6 +6,7 @@ import ( "app/base/models" "app/base/mqueue" "app/base/utils" + "context" "testing" "github.com/bytedance/sonic" @@ -28,7 +29,7 @@ func TestUpdateSystem(t *testing.T) { name := "TEST_NAME" ev.Host.DisplayName = &name ev.Host.SystemProfile.InstalledPackages = &[]string{"kernel-0:4.18.0-193.1.2.el8_2.x86_64"} - assert.NoError(t, HandleUpload(ev)) + assert.NoError(t, HandleUpload(context.Background(), ev)) var system models.SystemInventory assert.NoError(t, database.DB.Order("ID DESC").Find(&system, "inventory_id = ?", testInventoryID).Error) @@ -41,7 +42,7 @@ func TestDeleteSystem(t *testing.T) { createTestSystemInDB(t, testInventoryID, 1, testInventoryID.String()) deleteEvent := createTestDeleteEvent(testInventoryID) - err := HandleDelete(deleteEvent) + err := HandleDelete(context.Background(), deleteEvent) assert.NoError(t, err) assertSystemCulled(t) deleteData(t) @@ -56,7 +57,7 @@ func TestDeleteSystemWarn1(t *testing.T) { data, err := sonic.Marshal(deleteEvent) assert.NoError(t, err) - err = EventsMessageHandler(mqueue.KafkaMessage{Value: data}) + err = EventsMessageHandler(context.Background(), mqueue.KafkaMessage{Value: data}) assert.Equal(t, WarnEmptyEventType, logHook.LogEntries[len(logHook.LogEntries)-1].Message) assert.NoError(t, err) @@ -72,7 +73,7 @@ func TestDeleteSystemWarn2(t *testing.T) { data, err := sonic.Marshal(deleteEvent) assert.NoError(t, err) - err = EventsMessageHandler(mqueue.KafkaMessage{Value: data}) + err = EventsMessageHandler(context.Background(), mqueue.KafkaMessage{Value: data}) assert.Equal(t, WarnUnknownType, logHook.LogEntries[len(logHook.LogEntries)-1].Message) assert.NoError(t, err) @@ -87,7 +88,7 @@ func TestDeleteSystemWarn3(t *testing.T) { log.AddHook(logHook) deleteEvent := createTestDeleteEvent(notexistid) - err := HandleDelete(deleteEvent) + err := HandleDelete(context.Background(), deleteEvent) assert.NoError(t, err) assert.Equal(t, WarnNoRowsModified, logHook.LogEntries[len(logHook.LogEntries)-1].Message) @@ -101,12 +102,12 @@ func TestUploadAfterDelete(t *testing.T) { // system is not in database and the first event is delete deleteEvent := createTestDeleteEvent(testInventoryID) - err := HandleDelete(deleteEvent) + err := HandleDelete(context.Background(), deleteEvent) assert.NoError(t, err) // upload will be skipped and system won't be created uploadEvent := createTestUploadEvent("1", testInventoryID, "puptoo", true, false, "created") - err = HandleUpload(uploadEvent) + err = HandleUpload(context.Background(), uploadEvent) assert.NoError(t, err) assertSystemNotInDB(t) @@ -122,19 +123,19 @@ func TestCreateDeleteUpload(t *testing.T) { uploadEvent := createTestUploadEvent("1", testInventoryID, "puptoo", true, false, "created") originalName := "UPLOADED" uploadEvent.Host.DisplayName = &originalName - err := HandleUpload(uploadEvent) + err := HandleUpload(context.Background(), uploadEvent) assert.NoError(t, err) // delete marks the system but not physically delete it deleteEvent := createTestDeleteEvent(testInventoryID) - err = HandleDelete(deleteEvent) + err = HandleDelete(context.Background(), deleteEvent) assert.NoError(t, err) assertSystemCulled(t) // second upload of now deleted system should not change anything changedName := "UPDATED" uploadEvent.Host.DisplayName = &changedName - err = HandleUpload(uploadEvent) + err = HandleUpload(context.Background(), uploadEvent) assert.NoError(t, err) var system models.SystemInventory diff --git a/listener/template_advisories.go b/listener/template_advisories.go index 37fd0a698..a1e21f7d2 100644 --- a/listener/template_advisories.go +++ b/listener/template_advisories.go @@ -1,16 +1,18 @@ package listener import ( + "app/base/api" "app/base/content_sources" "app/base/database" "app/base/models" "app/base/utils" "context" + "net/http" + "github.com/pkg/errors" "github.com/redhatinsights/platform-go-middlewares/v2/identity" "gorm.io/gorm" "gorm.io/gorm/clause" - "net/http" ) // nolint: funlen @@ -188,8 +190,11 @@ func httpCallCSTemplateAdvisories(ctx context.Context, templateUUID string) ( if contentSourcesClient == nil { return nil, nil, errors.New("content sources client is nil") } - client := *contentSourcesClient - client.DefaultHeaders = map[string]string{"x-rh-identity": header} + client := &api.Client{ + HTTPClient: contentSourcesClient.HTTPClient, + Debug: contentSourcesClient.Debug, + DefaultHeaders: map[string]string{"x-rh-identity": header}, + } url := contentSourcesBaseURL + "/templates/" + templateUUID + "/advisories/ids" var resp content_sources.TemplateAdvisoryIDsResponse diff --git a/listener/template_test.go b/listener/template_test.go index d1dea888b..677d63e75 100644 --- a/listener/template_test.go +++ b/listener/template_test.go @@ -6,6 +6,7 @@ import ( "app/base/models" "app/base/mqueue" "app/base/utils" + "context" "errors" "fmt" "strings" @@ -80,7 +81,7 @@ func TestCreateTemplate(t *testing.T) { assert.Equal(t, 0, len(noTemplates)) // process message - err = TemplatesMessageHandler(mqueue.KafkaMessage{Value: msg}) + err = TemplatesMessageHandler(context.Background(), mqueue.KafkaMessage{Value: msg}) assert.Nil(t, err) // assert templates exist and have correct data @@ -99,7 +100,7 @@ func TestCreateTemplate(t *testing.T) { assert.Nil(t, err) // process update - err = TemplatesMessageHandler(mqueue.KafkaMessage{Value: msg}) + err = TemplatesMessageHandler(context.Background(), mqueue.KafkaMessage{Value: msg}) assert.Nil(t, err) // assert templates exist and have correct data @@ -113,7 +114,7 @@ func TestCreateTemplate(t *testing.T) { assert.Nil(t, err) // process update - err = TemplatesMessageHandler(mqueue.KafkaMessage{Value: msg}) + err = TemplatesMessageHandler(context.Background(), mqueue.KafkaMessage{Value: msg}) assert.Nil(t, err) // assert templates exist and have correct data @@ -138,7 +139,7 @@ func TestTemplateErrors(t *testing.T) { assert.Nil(t, err) // process message - err = TemplatesMessageHandler(mqueue.KafkaMessage{Value: msg}) + err = TemplatesMessageHandler(context.Background(), mqueue.KafkaMessage{Value: msg}) expectedErr := errors.New(`creating template from message: ` + `ERROR: invalid input syntax for type uuid: "not-an-uuid" (SQLSTATE 22P02)\n` + `creating template from message: ` + @@ -166,7 +167,7 @@ func TestTemplateEmptyDescription(t *testing.T) { assert.Nil(t, err) // process message - err = TemplatesMessageHandler(mqueue.KafkaMessage{Value: msg}) + err = TemplatesMessageHandler(context.Background(), mqueue.KafkaMessage{Value: msg}) assert.Nil(t, err) after := testTemplatesInDB(t) diff --git a/listener/templates.go b/listener/templates.go index 9dbc6307c..f3b8055dd 100644 --- a/listener/templates.go +++ b/listener/templates.go @@ -23,7 +23,7 @@ const ( TemplateEventCreate = "template-created" ) -func TemplatesMessageHandler(m mqueue.KafkaMessage) error { +func TemplatesMessageHandler(_ context.Context, m mqueue.KafkaMessage) error { eType, event, err := processTemplateEvent(m.Value) if err != nil { utils.LogError("err", err, "skipping template event") diff --git a/listener/upload.go b/listener/upload.go index eee481220..3cf207c0f 100644 --- a/listener/upload.go +++ b/listener/upload.go @@ -12,6 +12,7 @@ import ( "app/base/utils" "app/base/vmaas" "app/manager/middlewares" + "context" "crypto/sha256" "encoding/hex" "encoding/json" @@ -137,7 +138,7 @@ func (y *YumUpdates) GetBuiltPkgcache() bool { return y.BuiltPkgcache } -func HandleUpload(event HostEvent) error { +func HandleUpload(ctx context.Context, event HostEvent) error { tStart := time.Now() defer utils.ObserveSecondsSince(tStart, messageHandlingDuration.WithLabelValues(EventUpload)) @@ -165,7 +166,7 @@ func HandleUpload(event HostEvent) error { utils.LogError("err", err, "Could not get yum updates") } - sys, err := processUpload(&event.Host, yumUpdates) + sys, err := processUpload(ctx, &event.Host, yumUpdates) if err != nil { return handleListenerErrors(stdErrors.Join(ErrProcessUpload, err), &event, &ptEvent, tStart, ErrorStatus) } @@ -184,9 +185,9 @@ func HandleUpload(event HostEvent) error { ptEvent.StatusMsg = ProcessingStatus if event.Type == "created" { - createdEventsBuffer.bufferEvalEvents(sys.GetInventoryID(), sys.Inventory.RhAccountID, &ptEvent) + createdEventsBuffer.bufferEvalEvents(sys.GetInventoryID(), sys.Inventory.RhAccountID, &ptEvent, ctx) } else { - updatedEventsBuffer.bufferEvalEvents(sys.GetInventoryID(), sys.Inventory.RhAccountID, &ptEvent) + updatedEventsBuffer.bufferEvalEvents(sys.GetInventoryID(), sys.Inventory.RhAccountID, &ptEvent, ctx) } logAndObserve(UploadSuccess, ReceivedSuccess, &event, &ptEvent, tStart, SuccessStatus, false) return nil @@ -774,7 +775,7 @@ func processModules(systemProfile *inventory.SystemProfile) *[]vmaas.UpdatesV3Re } // We have received new upload, update stored host data, and re-evaluate the host against VMaaS -func processUpload(host *Host, yumUpdates *YumUpdates) (*models.SystemPlatformV2, error) { +func processUpload(ctx context.Context, host *Host, yumUpdates *YumUpdates) (*models.SystemPlatformV2, error) { tStart := time.Now() defer utils.ObserveSecondsSince(tStart, messagePartDuration.WithLabelValues("upload-processing")) // Ensure we have account stored @@ -805,7 +806,7 @@ func processUpload(host *Host, yumUpdates *YumUpdates) (*models.SystemPlatformV2 updatesReq.SetReleasever(releasever) } - tx := database.DB.WithContext(base.Context).Begin() + tx := database.DB.WithContext(ctx).Begin() defer tx.Rollback() var deleted models.DeletedSystem diff --git a/listener/upload_test.go b/listener/upload_test.go index b701c8f78..34f5706e2 100644 --- a/listener/upload_test.go +++ b/listener/upload_test.go @@ -221,7 +221,7 @@ func TestUploadHandlerCreatedSystem(t *testing.T) { repos := append(event.Host.SystemProfile.GetYumRepos(), inventory.YumRepo{ID: "epel", Enabled: true}) event.Host.SystemProfile.YumRepos = &repos - err := HandleUpload(event) + err := HandleUpload(context.Background(), event) assert.NoError(t, err) reporterID := 1 @@ -238,7 +238,7 @@ func TestUploadHandlerCreatedSystem(t *testing.T) { // Test that second upload did not cause re-evaluation logHook := utils.NewTestLogHook() log.AddHook(logHook) - err = HandleUpload(event) + err = HandleUpload(context.Background(), event) assert.NoError(t, err) assertInLogs(t, UploadSuccessNoEval, logHook.LogEntries...) assertSystemReposInDB(t, inv.ID, []string{"epel-8"}) @@ -253,7 +253,7 @@ func TestUploadHandlerWarn(t *testing.T) { logHook := utils.NewTestLogHook() log.AddHook(logHook) noPkgsEvent := createTestUploadEvent("1", testInventoryID, "puptoo", false, false, "created") - err := HandleUpload(noPkgsEvent) + err := HandleUpload(context.Background(), noPkgsEvent) if assert.Error(t, err) { assert.ErrorIs(t, err, ErrNoPackages) } @@ -266,7 +266,7 @@ func TestUploadHandlerWarnSkipReporter(t *testing.T) { logHook := utils.NewTestLogHook() log.AddHook(logHook) noPkgsEvent := createTestUploadEvent("1", testInventoryID, "yupana", false, false, "created") - err := HandleUpload(noPkgsEvent) + err := HandleUpload(context.Background(), noPkgsEvent) if assert.Error(t, err) { assert.ErrorIs(t, err, ErrReporter) } @@ -280,7 +280,7 @@ func TestUploadHandlerWarnSkipHostType(t *testing.T) { log.AddHook(logHook) event := createTestUploadEvent("1", testInventoryID, "puptoo", true, false, "created") event.Host.SystemProfile.HostType = "edge" - err := HandleUpload(event) + err := HandleUpload(context.Background(), event) if assert.Error(t, err) { assert.ErrorIs(t, err, ErrHostType) } @@ -295,7 +295,7 @@ func TestUploadHandlerError1(t *testing.T) { log.AddHook(logHook) event := createTestUploadEvent("1", testInventoryID, "puptoo", true, false, "created") *event.Host.OrgID = "" - err := HandleUpload(event) + err := HandleUpload(context.Background(), event) if assert.Error(t, err) { assert.ErrorIs(t, err, ErrNoAccountProvided) } @@ -320,7 +320,7 @@ func TestUploadHandlerError2(t *testing.T) { log.AddHook(logHook) _ = getOrCreateTestAccount(t) event := createTestUploadEvent("1", testInventoryID, "puptoo", true, false, "created") - err := HandleUpload(event) + err := HandleUpload(context.Background(), event) assert.Nil(t, err) time.Sleep(2 * uploadEvalTimeout) assertInLogs(t, ErrorKafkaSend, logHook.LogEntries...) diff --git a/manager/manager.go b/manager/manager.go index fd875719e..3b0d4c442 100644 --- a/manager/manager.go +++ b/manager/manager.go @@ -4,6 +4,7 @@ import ( "app/base" "app/base/core" "app/base/mqueue" + "app/base/telemetry" "app/base/utils" "app/docs" "app/manager/config" @@ -43,6 +44,7 @@ func RunManager() { // middlewares app.Use(gin.Recovery()) + app.Use(telemetry.RHHTTPAttributes()) middlewares.Prometheus().Use(app) app.Use(middlewares.MaxConnections(utils.CoreCfg.MaxGinConnections)) app.Use(middlewares.Ratelimit(utils.CoreCfg.Ratelimit)) diff --git a/manager/middlewares/logger.go b/manager/middlewares/logger.go index 939ae75cf..02f817c35 100644 --- a/manager/middlewares/logger.go +++ b/manager/middlewares/logger.go @@ -35,9 +35,9 @@ func RequestResponseLogger() gin.HandlerFunc { fields = append(fields, "request") if c.Writer.Status() < http.StatusInternalServerError { - utils.LogInfo(fields...) + utils.LogInfo(append([]any{c.Request.Context()}, fields...)...) } else { - utils.LogError(fields...) + utils.LogError(append([]any{c.Request.Context()}, fields...)...) } utils.ObserveSecondsSince(tStart, requestDurations. diff --git a/manager/middlewares/rbac.go b/manager/middlewares/rbac.go index 753774edb..4ef373805 100644 --- a/manager/middlewares/rbac.go +++ b/manager/middlewares/rbac.go @@ -4,12 +4,14 @@ import ( "app/base" "app/base/api" "app/base/rbac" + "app/base/telemetry" "app/base/utils" "app/manager/config" "fmt" "net/http" "strconv" "strings" + "sync" "github.com/gin-gonic/gin" "github.com/google/uuid" @@ -22,6 +24,7 @@ const rootWorkspaceID = "00000000-0000-0000-0000-999999999999" var ( rbacURL = "" httpClient = &http.Client{} + httpOnce sync.Once ) const xRHIdentity = "x-rh-identity" @@ -39,6 +42,9 @@ var granularPerms = map[string]string{ // Make RBAC client on demand, with specified identity func makeClient(identity string) *api.Client { + httpOnce.Do(func() { + httpClient = telemetry.InstrumentHTTPClient(httpClient) + }) debugRequest := log.IsLevelEnabled(log.TraceLevel) client := api.Client{