Feat: update the pr as per comment

Author: Manendra Pal Singh

cmd/adapter/main.go

@@ -233,7 +233,7 @@ func addParentIdCtx(ctx context.Context, config *Config) context.Context {
 		log.Infof(ctx, "Adding POD name: %s", p)
 		podName = p
 	} else {
-		log.Info(ctx, "POD_NAME environment variable not set falling back to hostname")
+		log.Info(ctx, "POD_NAME environment variable not set, falling back to hostname")
 		if hostname, err := os.Hostname(); err == nil {
 			log.Infof(ctx, "Setting POD name as hostname: %s", hostname)
 			podName = hostname
@@ -243,16 +243,21 @@ func addParentIdCtx(ctx context.Context, config *Config) context.Context {
 	}
 
 	for _, m := range config.Modules {
-		if m.Handler.Role != "" && m.Handler.SubscriberID != "" {
-			parentID = string(m.Handler.Role) + ":" + m.Handler.SubscriberID + ":" + podName
-			break
+		if m.Handler.Role == "" || m.Handler.SubscriberID == "" {
+			continue
+		}
+		candidate := string(m.Handler.Role) + ":" + m.Handler.SubscriberID + ":" + podName
+		if parentID == "" {
+			parentID = candidate
+		} else if candidate != parentID {
+			log.Warnf(ctx, "Multiple distinct role:subscriberID pairs found in modules (using %q, also saw %q); consider explicit parent_id config", parentID, candidate)
 		}
 	}
 
 	if parentID != "" {
 		ctx = context.WithValue(ctx, model.ContextKeyParentID, parentID)
 	} else {
-		log.Warnf(ctx, "Failed to find parent ID in config please add the role and subscriber_id in the handler config ")
+		log.Warnf(ctx, "Failed to find parent ID in config; add role and subscriber_id to the handler config")
 	}
 	return ctx
 }

core/module/handler/handlerMetrics.go

@@ -65,4 +65,3 @@ func newHandlerMetrics() (*HandlerMetrics, error) {
 
 	return m, nil
 }
-

core/module/handler/http_metric.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"strings"
 	"sync"
 
 	"github.com/beckn-one/beckn-onix/pkg/telemetry"
@@ -79,10 +80,10 @@ func RecordHTTPRequest(ctx context.Context, statusCode int, action, role, sender
 
 	metric_code := action + "_api_total_count"
 	category := "NetworkHealth"
-	if action == "/search" || action == "/discovery" {
+	if strings.HasSuffix(action, "/search") || strings.HasSuffix(action, "/discovery") {
 		category = "Discovery"
 	}
-	attributes = append(attributes, specHttpMetricAttr(metric_code, category)...) //TODO: need to update as per the furthur discussion
+	attributes = append(attributes, specHttpMetricAttr(metric_code, category)...)
 	m.HttpRequestCount.Add(ctx, 1, metric.WithAttributes(attributes...))
 }
 

core/module/handler/stdHandler.go

@@ -122,19 +122,7 @@ func (h *stdHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		record:         nil,
 	}
 
-	selfID := h.SubscriberID
-	remoteID := ""
-	if v, ok := r.Context().Value(model.ContextKeyRemoteID).(string); ok {
-		remoteID = v
-	}
-	var senderID, receiverID string
-	if strings.Contains(h.moduleName, "Caller") {
-		senderID = selfID
-		receiverID = remoteID
-	} else {
-		senderID = remoteID
-		receiverID = selfID
-	}
+	senderID, receiverID := h.resolveDirection(r.Context())
 	httpMeter, _ := GetHTTPMetrics(r.Context())
 	if httpMeter != nil {
 		recordOnce = func() {
@@ -161,7 +149,7 @@ func (h *stdHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 
 		body := stepCtx.Body
-		go telemetry.EmitAuditLogs(r.Context(), body, auditlog.Int("http.response.status_code", wrapped.statusCode), auditlog.String("http.response.error", errString(err)))
+		telemetry.EmitAuditLogs(r.Context(), body, auditlog.Int("http.response.status_code", wrapped.statusCode), auditlog.String("http.response.error", errString(err)))
 		span.End()
 	}()
 
@@ -385,21 +373,17 @@ func (h *stdHandler) initSteps(ctx context.Context, mgr PluginManager, cfg *Conf
 	return nil
 }
 
-func setBecknAttr(span trace.Span, r *http.Request, h *stdHandler) {
+func (h *stdHandler) resolveDirection(ctx context.Context) (senderID, receiverID string) {
 	selfID := h.SubscriberID
-	remoteID := ""
-	if v, ok := r.Context().Value(model.ContextKeyRemoteID).(string); ok {
-		remoteID = v
-	}
-
-	var senderID, receiverID string
+	remoteID, _ := ctx.Value(model.ContextKeyRemoteID).(string)
 	if strings.Contains(h.moduleName, "Caller") {
-		senderID = selfID
-		receiverID = remoteID
-	} else {
-		senderID = remoteID
-		receiverID = selfID
+		return selfID, remoteID
 	}
+	return remoteID, selfID
+}
+
+func setBecknAttr(span trace.Span, r *http.Request, h *stdHandler) {
+	senderID, receiverID := h.resolveDirection(r.Context())
 	attrs := []attribute.KeyValue{
 		telemetry.AttrRecipientID.String(receiverID),
 		telemetry.AttrSenderID.String(senderID),

core/module/handler/step.go

@@ -41,28 +41,25 @@ func (s *signStep) Run(ctx *model.StepContext) error {
 		return model.NewBadReqErr(fmt.Errorf("subscriberID not set"))
 	}
 
-	tracer := otel.Tracer("beckn-onix")
+	tracer := otel.Tracer(telemetry.ScopeName, trace.WithInstrumentationVersion(telemetry.ScopeVersion))
 
 	var keySet *model.Keyset
 	{
-		// to create span to finding the key set
 		keySetCtx, keySetSpan := tracer.Start(ctx.Context, "keyset")
-		defer keySetSpan.End()
 		ks, err := s.km.Keyset(keySetCtx, ctx.SubID)
+		keySetSpan.End()
 		if err != nil {
 			return fmt.Errorf("failed to get signing key: %w", err)
 		}
 		keySet = ks
-
 	}
 
 	{
-		// to create span for the signa
 		signerCtx, signerSpan := tracer.Start(ctx.Context, "sign")
-		defer signerSpan.End()
 		createdAt := time.Now().Unix()
 		validTill := time.Now().Add(5 * time.Minute).Unix()
 		sign, err := s.signer.Sign(signerCtx, ctx.Body, keySet.SigningPrivate, createdAt, validTill)
+		signerSpan.End()
 		if err != nil {
 			return fmt.Errorf("failed to sign request: %w", err)
 		}
@@ -73,7 +70,6 @@ func (s *signStep) Run(ctx *model.StepContext) error {
 			header = model.AuthHeaderGateway
 		}
 		ctx.Request.Header.Set(header, authHeader)
-
 	}
 
 	return nil

install/network-observability/docker-compose.yml

@@ -196,46 +196,41 @@ services:
       - zipkin
       - loki
 
-  bpp-client:
-    image: fidedocker/protocol-server
-    container_name: bpp-client
+  sandbox-bap:
+    container_name: sandbox-bap
+    image: fidedocker/sandbox-2.0:latest
     platform: linux/amd64
-    networks:
-      - beckn_network
+    environment:
+      - NODE_ENV=production
+      - PORT=3001
     ports:
-      - "6001:6001"
-    restart: unless-stopped
-    volumes:
-      - bpp_client_config_volume:/usr/src/app/config
-      - bpp_client_schemas_volume:/usr/src/app/schemas
-      - bpp_client_logs_volume:/usr/src/app/logs
-
-  bpp-network:
-    image: fidedocker/protocol-server
-    container_name: bpp-network
-    platform: linux/amd64
+      - "3001:3001"
+    healthcheck:
+      test: ["CMD", "wget", "-qO-", "http://localhost:3001/api/health"]
+      interval: 10s
+      timeout: 3s
+      retries: 5
+      start_period: 10s
     networks:
       - beckn_network
-    ports:
-      - "6002:6002"
-    restart: unless-stopped
-    volumes:
-      - bpp_network_config_volume:/usr/src/app/config
-      - bpp_network_schemas_volume:/usr/src/app/schemas
-      - bpp_network_logs_volume:/usr/src/app/logs
 
-  sandbox-api:
-    image: fidedocker/sandbox-api
-    container_name: sandbox-api
+  sandbox-bpp:
+    container_name: sandbox-bpp
+    image: fidedocker/sandbox-2.0:latest
     platform: linux/amd64
+    environment:
+      - NODE_ENV=production
+      - PORT=3002
+    ports:
+      - "3002:3002"
+    healthcheck:
+      test: ["CMD", "wget", "-qO-", "http://localhost:3002/api/health"]
+      interval: 10s
+      timeout: 3s
+      retries: 5
+      start_period: 10s
     networks:
       - beckn_network
-    ports:
-      - "4010:4000"
-    restart: unless-stopped
-    environment:
-      - PORT=4000
-      - WEBHOOK_URL=http://host.docker.internal:3001/webhook
 
 networks:
   observability:
@@ -248,17 +243,3 @@ volumes:
   prometheus_data:
   grafana_data:
   loki_data:
-  bpp_client_config_volume:
-    name: bpp_client_config_volume
-    external: true
-  bpp_client_schemas_volume:
-    name: bpp_client_schemas_volume
-  bpp_client_logs_volume:
-    name: bpp_client_logs_volume
-  bpp_network_config_volume:
-    name: bpp_network_config_volume
-    external: true
-  bpp_network_schemas_volume:
-    name: bpp_network_schemas_volume
-  bpp_network_logs_volume:
-    name: bpp_network_logs_volume

pkg/log/log_test.go

@@ -250,11 +250,11 @@ func TestError(t *testing.T) {
 func TestRequest(t *testing.T) {
 	logPath := setupLogger(t, InfoLevel)
 	ctx := context.WithValue(context.Background(), requestID, "abc-123")
-	ctx = context.WithValue(context.Background(), transaction_id, "transaction-id-123-")
-	ctx = context.WithValue(context.Background(), message_id, "message-id-123")
-	ctx = context.WithValue(context.Background(), subscriber_id, "subscriber-id-123")
-	ctx = context.WithValue(context.Background(), module_id, "module-id-123")
-	ctx = context.WithValue(context.Background(), parent_id, "parent-id-123")
+	ctx = context.WithValue(ctx, transaction_id, "transaction-id-123-")
+	ctx = context.WithValue(ctx, message_id, "message-id-123")
+	ctx = context.WithValue(ctx, subscriber_id, "subscriber-id-123")
+	ctx = context.WithValue(ctx, module_id, "module-id-123")
+	ctx = context.WithValue(ctx, parent_id, "parent-id-123")
 
 	req, _ := http.NewRequest("POST", "/api/test", bytes.NewBuffer([]byte(`{"key":"value"}`)))
 	req.RemoteAddr = "127.0.0.1:8080"

pkg/model/model.go

@@ -58,7 +58,7 @@ const (
 	ContextKeyParentID ContextKey = "parent_id"
 
 	// ContextKeyRemoteID is the context key for the caller who is calling the bap/bpp
-	ContextKeyRemoteID ContextKey = "caller_id"
+	ContextKeyRemoteID ContextKey = "remote_id"
 )
 
 var contextKeys = map[string]ContextKey{
@@ -67,7 +67,7 @@ var contextKeys = map[string]ContextKey{
 	"subscriber_id":  ContextKeySubscriberID,
 	"module_id":      ContextKeyModuleID,
 	"parent_id":      ContextKeyParentID,
-	"caller_id":      ContextKeyRemoteID,
+	"remote_id":      ContextKeyRemoteID,
 }
 
 // ParseContextKey converts a string into a valid ContextKey.

pkg/plugin/implementation/cache/cache.go

@@ -114,22 +114,25 @@ func New(ctx context.Context, cfg *Config) (*Cache, func() error, error) {
 
 // Get retrieves the value for the specified key from Redis.
 func (c *Cache) Get(ctx context.Context, key string) (string, error) {
-	result, err := c.Client.Get(ctx, key).Result()
+	tracer := otel.Tracer(telemetry.ScopeName, trace.WithInstrumentationVersion(telemetry.ScopeVersion))
+	spanCtx, span := tracer.Start(ctx, "redis_get")
+	defer span.End()
+	result, err := c.Client.Get(spanCtx, key).Result()
 	if c.metrics != nil {
 		attrs := []attribute.KeyValue{
 			telemetry.AttrOperation.String("get"),
 		}
 		switch {
 		case err == redis.Nil:
-			c.metrics.CacheMissesTotal.Add(ctx, 1, metric.WithAttributes(attrs...))
-			c.metrics.CacheOperationsTotal.Add(ctx, 1,
+			c.metrics.CacheMissesTotal.Add(spanCtx, 1, metric.WithAttributes(attrs...))
+			c.metrics.CacheOperationsTotal.Add(spanCtx, 1,
 				metric.WithAttributes(append(attrs, telemetry.AttrStatus.String("miss"))...))
 		case err != nil:
-			c.metrics.CacheOperationsTotal.Add(ctx, 1,
+			c.metrics.CacheOperationsTotal.Add(spanCtx, 1,
 				metric.WithAttributes(append(attrs, telemetry.AttrStatus.String("error"))...))
 		default:
-			c.metrics.CacheHitsTotal.Add(ctx, 1, metric.WithAttributes(attrs...))
-			c.metrics.CacheOperationsTotal.Add(ctx, 1,
+			c.metrics.CacheHitsTotal.Add(spanCtx, 1, metric.WithAttributes(attrs...))
+			c.metrics.CacheOperationsTotal.Add(spanCtx, 1,
 				metric.WithAttributes(append(attrs, telemetry.AttrStatus.String("hit"))...))
 		}
 	}
@@ -149,8 +152,11 @@ func (c *Cache) Set(ctx context.Context, key, value string, ttl time.Duration) e
 
 // Delete removes the specified key from Redis.
 func (c *Cache) Delete(ctx context.Context, key string) error {
-	err := c.Client.Del(ctx, key).Err()
-	c.recordOperation(ctx, "delete", err)
+	tracer := otel.Tracer(telemetry.ScopeName, trace.WithInstrumentationVersion(telemetry.ScopeVersion))
+	spanCtx, span := tracer.Start(ctx, "redis_delete")
+	defer span.End()
+	err := c.Client.Del(spanCtx, key).Err()
+	c.recordOperation(spanCtx, "delete", err)
 	return err
 }