RHINENG-28312: Implement OpenTelemetry

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")

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

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)

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")
 	}()
 }

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()
 }

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)
 	}

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")

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

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")
 }

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,13 +34,15 @@ type EvalData struct {
 	RhAccountID int
 	RequestID   string
 	OrgID       *string
+	Traceparent string
 }
 
 type PlatformEvents []PlatformEvent
 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,
 			})

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"