Multiple Improvements and Bug Fixes Related to Analytics Flow

gateway/build-manifest.yaml

@@ -1,7 +1,7 @@
 version: v1
 policies:
     - name: advanced-ratelimit
-      version: v1.0.3
+      version: v1.1.0
       gomodule: github.com/wso2/gateway-controllers/policies/advanced-ratelimit@v1
     - name: analytics-header-filter
       version: v1.0.1
@@ -22,7 +22,7 @@ policies:
       version: v1.0.2
       gomodule: github.com/wso2/gateway-controllers/policies/basic-auth@v1
     - name: basic-ratelimit
-      version: v1.0.2
+      version: v1.1.0
       gomodule: github.com/wso2/gateway-controllers/policies/basic-ratelimit@v1
     - name: content-length-guardrail
       version: v1.0.1
@@ -52,7 +52,7 @@ policies:
       version: v1.0.3
       gomodule: github.com/wso2/gateway-controllers/policies/llm-cost@v1
     - name: llm-cost-based-ratelimit
-      version: v1.0.3
+      version: v1.1.0
       gomodule: github.com/wso2/gateway-controllers/policies/llm-cost-based-ratelimit@v1
     - name: log-message
       version: v1.0.2
@@ -67,7 +67,7 @@ policies:
       version: v1.0.1
       gomodule: github.com/wso2/gateway-controllers/policies/mcp-authz@v1
     - name: mcp-ratelimit
-      version: v1.0.0
+      version: v1.1.0
       gomodule: github.com/wso2/gateway-controllers/policies/mcp-ratelimit@v1
     - name: mcp-rewrite
       version: v1.0.2
@@ -121,7 +121,7 @@ policies:
       version: v1.0.2
       gomodule: github.com/wso2/gateway-controllers/policies/subscription-validation@v1
     - name: token-based-ratelimit
-      version: v1.0.3
+      version: v1.1.0
       gomodule: github.com/wso2/gateway-controllers/policies/token-based-ratelimit@v1
     - name: url-guardrail
       version: v1.0.1

gateway/gateway-runtime/policy-engine/internal/analytics/analytics.go

@@ -415,15 +415,26 @@ func (c *Analytics) prepareAnalyticEvent(logEntry *v3.HTTPAccessLogEntry) *dto.E
 	event.Properties["userName"] = userName
 	event.Properties["commonName"] = "N/A"
 	event.Properties["apiContext"] = extendedAPI.APIContext
-	if logEntry.Response != nil {
+	// Resolve responseContentType for all API kinds. The analytics system policy captures
+	// it from the response headers into analytics metadata (response_content_type) because
+	// the Envoy access log does not carry response headers. Prefer that value; fall back to
+	// the access-log header if present, then to Unknown.
+	responseContentType := Unknown
+	if ct, ok := keyValuePairsFromMetadata["response_content_type"]; ok && ct != "" {
+		responseContentType = ct
+	} else if logEntry.Response != nil {
 		if contentTypeHeader := logEntry.Response.GetResponseHeaders()["content-type"]; contentTypeHeader != "" {
-			event.Properties["responseContentType"] = contentTypeHeader
-		} else {
-			event.Properties["responseContentType"] = Unknown
+			responseContentType = contentTypeHeader
 		}
+	}
+	event.Properties["responseContentType"] = responseContentType
+	if logEntry.Response != nil {
 		event.Properties["responseSize"] = logEntry.Response.ResponseBodyBytes
-	} else {
-		event.Properties["responseContentType"] = Unknown
+	}
+
+	// requestSize is common to all API kinds; mirror responseSize using the Envoy access-log byte count.
+	if request != nil {
+		event.Properties["requestSize"] = request.GetRequestBodyBytes()
 	}
 
 	//Adding request and response headers for the analytics event
@@ -451,7 +462,7 @@ func (c *Analytics) prepareAnalyticEvent(logEntry *v3.HTTPAccessLogEntry) *dto.E
 	if keyValuePairsFromMetadata[APITypeKey] != "" && keyValuePairsFromMetadata[APITypeKey] == "Mcp" {
 		mcpAnalytics := make(map[string]interface{})
 		if mcpSessionID, ok := keyValuePairsFromMetadata["mcp_session_id"]; ok && mcpSessionID != "" {
-			mcpAnalytics["mcp_session_id"] = mcpSessionID
+			mcpAnalytics["sessionId"] = mcpSessionID
 		}
 		if mcpRequestProps, ok := keyValuePairsFromMetadata["mcp_request_properties"]; ok && mcpRequestProps != "" {
 			// Parse the JSON string into a map

gateway/gateway-runtime/policy-engine/internal/analytics/analytics_test.go

@@ -711,7 +711,39 @@ func TestPrepareAnalyticEvent_WithMCPAnalytics(t *testing.T) {
 
 	mcpMap, ok := mcpAnalytics.(map[string]interface{})
 	require.True(t, ok)
-	assert.Equal(t, "session-123", mcpMap["mcp_session_id"])
+	assert.Equal(t, "session-123", mcpMap["sessionId"])
+}
+
+func TestPrepareAnalyticEvent_MCPCapabilityContentTypeAndSizes(t *testing.T) {
+	cfg := &config.Config{}
+	analytics := NewAnalytics(cfg)
+
+	logEntry := createLogEntryWithMetadata(map[string]string{
+		APITypeKey:               "Mcp",
+		"mcp_session_id":         "session-123",
+		"mcp_request_properties": `{"jsonRpcMethod":"tools/call","capability":"TOOL","capabilityName":"calculator"}`,
+		"response_content_type":  "text/event-stream",
+	})
+	// requestSize is sourced from the Envoy access-log request body byte count.
+	logEntry.Request.RequestBodyBytes = 42
+
+	event := analytics.prepareAnalyticEvent(logEntry)
+	require.NotNil(t, event)
+
+	// requestSize is published at the root level for all API kinds.
+	assert.Equal(t, uint64(42), event.Properties["requestSize"])
+	// responseContentType is captured from the system policy's response headers
+	// and published at the root level (not duplicated inside mcpAnalytics).
+	assert.Equal(t, "text/event-stream", event.Properties["responseContentType"])
+
+	mcpAnalytics, ok := event.Properties["mcpAnalytics"]
+	require.True(t, ok)
+	mcpMap, ok := mcpAnalytics.(map[string]interface{})
+	require.True(t, ok)
+
+	// capability/capabilityName flow through from mcp_request_properties.
+	assert.Equal(t, "TOOL", mcpMap["capability"])
+	assert.Equal(t, "calculator", mcpMap["capabilityName"])
 }
 
 func TestPrepareAnalyticEvent_WithMCPAnalyticsInvalidJSON(t *testing.T) {
@@ -764,6 +796,40 @@ func TestPrepareAnalyticEvent_WithResponseContentType(t *testing.T) {
 	assert.Equal(t, "application/json", event.Properties["responseContentType"])
 }
 
+// The analytics system policy captures response_content_type into metadata for all API
+// kinds (the Envoy access log carries no response headers). Verify a non-MCP (REST) API
+// sources responseContentType from that metadata rather than falling back to Unknown.
+func TestPrepareAnalyticEvent_ResponseContentTypeFromMetadataNonMCP(t *testing.T) {
+	cfg := &config.Config{}
+	analytics := NewAnalytics(cfg)
+
+	logEntry := createLogEntryWithMetadata(map[string]string{
+		APITypeKey:              "Rest",
+		"response_content_type": "application/json",
+	})
+
+	event := analytics.prepareAnalyticEvent(logEntry)
+
+	require.NotNil(t, event)
+	assert.Equal(t, "application/json", event.Properties["responseContentType"])
+}
+
+// When neither metadata nor the access-log header carries a content type, the value
+// falls back to Unknown.
+func TestPrepareAnalyticEvent_ResponseContentTypeFallbackUnknown(t *testing.T) {
+	cfg := &config.Config{}
+	analytics := NewAnalytics(cfg)
+
+	logEntry := createLogEntryWithMetadata(map[string]string{
+		APITypeKey: "Rest",
+	})
+
+	event := analytics.prepareAnalyticEvent(logEntry)
+
+	require.NotNil(t, event)
+	assert.Equal(t, Unknown, event.Properties["responseContentType"])
+}
+
 func TestPrepareAnalyticEvent_WithCorrelationID(t *testing.T) {
 	cfg := &config.Config{}
 	analytics := NewAnalytics(cfg)

gateway/gateway-runtime/policy-engine/internal/analytics/publishers/moesif.go

@@ -134,57 +134,31 @@ func (m *Moesif) Publish(event *dto.Event) {
 		uri = event.Operation.APIResourceTemplate
 	}
 
-	// Build request headers: prefer dynamic headers from event.Properties["requestHeaders"]
-	// if present; otherwise, fall back to the existing hardcoded headers.
-	defaultReqHeaders := map[string]interface{}{
-		"User-Agent":   event.UserAgentHeader,
-		"Content-Type": "-",
-	}
-
-	defaultRspHeaders := map[string]interface{}{
-		"Vary":          "Accept-Encoding",
-		"Pragma":        "no-cache",
-		"Expires":       "-1",
-		"Content-Type":  "-",
-		"Cache-Control": "no-cache",
-	}
-
-	headers := defaultReqHeaders
+	// Headers are only published when the analytics-header-filter policy is configured
+	// and has emitted the (already allow/deny filtered) header set into event metadata.
+	// When it is not configured, no headers are sent at all.
+	headers := map[string]interface{}{}
 	if rawReqHeaders, ok := event.Properties["requestHeaders"]; ok && rawReqHeaders != nil {
 		slog.Debug("Request headers (PUBLISHER): ", "requestHeaders", rawReqHeaders)
 		if jsonStr, ok := rawReqHeaders.(string); ok {
 			var hMap map[string]interface{}
 			if err := json.Unmarshal([]byte(jsonStr), &hMap); err == nil && len(hMap) > 0 {
 				slog.Debug("Unmarshalled hMap (PUBLISHER): ", "requestHeaders", hMap)
-				merged := make(map[string]interface{})
-				for k, v := range defaultReqHeaders {
-					merged[k] = v
-				}
-				for k, v := range hMap {
-					merged[k] = v
-				}
-				headers = merged
+				headers = hMap
 			} else if err != nil {
 				slog.Warn("Failed to unmarshal request headers", "error", err)
 			}
 		}
 	}
 
-	rspHeaders := defaultRspHeaders
+	rspHeaders := map[string]interface{}{}
 	if rawRspHeaders, ok := event.Properties["responseHeaders"]; ok && rawRspHeaders != nil {
 		slog.Debug("Response headers (PUBLISHER): ", "responseHeaders", rawRspHeaders)
 		if jsonStr, ok := rawRspHeaders.(string); ok {
 			var hMap map[string]interface{}
 			if err := json.Unmarshal([]byte(jsonStr), &hMap); err == nil && len(hMap) > 0 {
 				slog.Debug("Unmarshalled hMap (PUBLISHER): ", "responseHeaders", hMap)
-				merged := make(map[string]interface{})
-				for k, v := range defaultRspHeaders {
-					merged[k] = v
-				}
-				for k, v := range hMap {
-					merged[k] = v
-				}
-				rspHeaders = merged
+				rspHeaders = hMap
 			} else if err != nil {
 				slog.Warn("Failed to unmarshal response headers", "error", err)
 			}
@@ -276,6 +250,11 @@ func (m *Moesif) Publish(event *dto.Event) {
 		metadataMap["responseSize"] = responseSize
 	}
 
+	// requestSize
+	if requestSize, ok := event.Properties["requestSize"]; ok && requestSize != nil {
+		metadataMap["requestSize"] = requestSize
+	}
+
 	// Advanced latency info
 	if event.Latencies != nil {
 		metadataMap["backendLatency"] = event.Latencies.BackendLatency

gateway/gateway-runtime/policy-engine/internal/analytics/publishers/moesif_test.go

@@ -175,9 +175,9 @@ func TestPublish_WithInvalidRequestHeaders(t *testing.T) {
 	moesif.Publish(event)
 
 	assert.Len(t, moesif.events, 1)
-	// Should fall back to default headers
+	// Invalid JSON: no headers configured -> no headers published (no defaults).
 	headers := moesif.events[0].Request.Headers.(map[string]interface{})
-	assert.Equal(t, "test-agent", headers["User-Agent"])
+	assert.Empty(t, headers)
 }
 
 func TestPublish_WithEmptyRequestHeaders(t *testing.T) {
@@ -189,9 +189,9 @@ func TestPublish_WithEmptyRequestHeaders(t *testing.T) {
 	moesif.Publish(event)
 
 	assert.Len(t, moesif.events, 1)
-	// Empty JSON object should fall back to default headers
+	// Empty JSON object -> no headers published (no defaults).
 	headers := moesif.events[0].Request.Headers.(map[string]interface{})
-	assert.Equal(t, "test-agent", headers["User-Agent"])
+	assert.Empty(t, headers)
 }
 
 func TestPublish_WithResponseHeaders(t *testing.T) {
@@ -217,9 +217,25 @@ func TestPublish_WithInvalidResponseHeaders(t *testing.T) {
 	moesif.Publish(event)
 
 	assert.Len(t, moesif.events, 1)
-	// Should fall back to default headers
+	// Invalid JSON: no headers configured -> no headers published (no defaults).
 	headers := moesif.events[0].Response.Headers.(map[string]interface{})
-	assert.Equal(t, "no-cache", headers["Cache-Control"])
+	assert.Empty(t, headers)
+}
+
+func TestPublish_NoHeadersConfigured(t *testing.T) {
+	moesif := createTestMoesifWithoutAPI()
+
+	event := createBaseEvent()
+	// No requestHeaders/responseHeaders in event properties (analytics-header-filter
+	// policy not configured) -> both header sets must be empty.
+
+	moesif.Publish(event)
+
+	assert.Len(t, moesif.events, 1)
+	reqHeaders := moesif.events[0].Request.Headers.(map[string]interface{})
+	rspHeaders := moesif.events[0].Response.Headers.(map[string]interface{})
+	assert.Empty(t, reqHeaders)
+	assert.Empty(t, rspHeaders)
 }
 
 func TestPublish_LlmProviderWithAIMetadata(t *testing.T) {
@@ -424,9 +440,9 @@ func TestPublish_RequestHeadersNonString(t *testing.T) {
 	moesif.Publish(event)
 
 	assert.Len(t, moesif.events, 1)
-	// Should use default headers
+	// Non-string value -> no headers configured -> no headers published (no defaults).
 	headers := moesif.events[0].Request.Headers.(map[string]interface{})
-	assert.Equal(t, "test-agent", headers["User-Agent"])
+	assert.Empty(t, headers)
 }
 
 func TestPublish_ResponseHeadersNonString(t *testing.T) {
@@ -439,9 +455,9 @@ func TestPublish_ResponseHeadersNonString(t *testing.T) {
 	moesif.Publish(event)
 
 	assert.Len(t, moesif.events, 1)
-	// Should use default headers
+	// Non-string value -> no headers configured -> no headers published (no defaults).
 	headers := moesif.events[0].Response.Headers.(map[string]interface{})
-	assert.Equal(t, "no-cache", headers["Cache-Control"])
+	assert.Empty(t, headers)
 }
 
 func TestPublish_MetadataContainsAPIInfo(t *testing.T) {

gateway/system-policies/analytics/analytics.go

@@ -192,6 +192,15 @@ func (a *AnalyticsPolicy) OnResponseHeaders(_ context.Context, respCtx *policy.R
 		}
 	}
 
+	// Capture the response content type for all API kinds. The Envoy access log does
+	// not carry response headers (no additional_response_headers_to_log is configured,
+	// to avoid an uncontrolled header source)
+	if respCtx.ResponseHeaders != nil {
+		if contentTypes := respCtx.ResponseHeaders.Get("content-type"); len(contentTypes) > 0 {
+			analyticsMetadata["response_content_type"] = contentTypes[0]
+		}
+	}
+
 	if respCtx.SharedContext.APIKind == policy.APIKindMCP && respCtx.ResponseHeaders != nil {
 		if sessionIDs := respCtx.ResponseHeaders.Get("mcp-session-id"); len(sessionIDs) > 0 {
 			analyticsMetadata["mcp_session_id"] = sessionIDs[0]
@@ -255,7 +264,7 @@ func (a *AnalyticsPolicy) OnRequestBody(_ context.Context, ctx *policy.RequestCo
 
 			props.JsonRpcMethod = extractString(JsonRpcMethodJsonPath)
 			props.CapabilityName = extractString(McpCapabilityNameJsonPath)
-			props.Capability = extractString(McpResourceUriJsonPath)
+			props.Capability = deriveMCPCapability(props.JsonRpcMethod)
 
 			clientInfo := McpClientInfo{
 				RequestedProtocolVersion: extractStringFromJsonpath(mcpPayload, ProtocolVersionJsonPath),
@@ -757,10 +766,16 @@ func extractMCPResponseAnalyticsProps(payload map[string]interface{}) *McpRespon
 		props.ServerInfo = &serverInfo
 	}
 
-	isError, err := extractBoolFromJsonpath(payload, IsErrorJsonPath)
-	if err == nil {
-		props.IsError = &isError
+	// isError denotes whether the JSON-RPC response represents an error. It is true when a
+	// protocol-level error object is present ($.error) or when a tool result explicitly sets
+	// result.isError=true; false otherwise. Always emitted so consumers get a definitive flag.
+	isError := false
+	if errVal, hasError := payload["error"]; hasError && errVal != nil {
+		isError = true
+	} else if resultIsError, err := extractBoolFromJsonpath(payload, IsErrorJsonPath); err == nil {
+		isError = resultIsError
 	}
+	props.IsError = &isError
 
 	errorCode, err := extractIntFromJsonpath(payload, JsonRpcErrorCodeJsonPath)
 	if err == nil {
@@ -968,3 +983,19 @@ func extractBoolFromJsonpath(payload map[string]interface{}, path string) (bool,
 		return false, fmt.Errorf("unexpected type %T at path %s", val, path)
 	}
 }
+
+// deriveMCPCapability maps an MCP JSON-RPC method to a capability type for analytics.
+// It returns "" when the method has no mapping, in which case the caller omits the
+// field (mirrors carbon-apimgt's SynapseAnalyticsDataProvider behavior).
+func deriveMCPCapability(method string) string {
+	switch {
+	case strings.HasPrefix(method, "tools/"):
+		return "TOOL"
+	case strings.HasPrefix(method, "resources/"):
+		return "RESOURCE"
+	case strings.HasPrefix(method, "prompts/"):
+		return "PROMPT"
+	default:
+		return ""
+	}
+}