Emit metrics from potentially changed header handlers

CHANGELOG.md

@@ -8,6 +8,18 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ## [Unreleased]
 - Upgraded go version to 1.21, set toolchain version.
 - Reverted rpc-caller-procedure value setting.
+- Preparation for the new header handling behavior.
+
+Starting from one of the next releases, following behaviour changes will be applied:
+- Any inbound header with the prefix `rpc-` will be treated as a reserved header and will be ignored
+(i.e. not forwarded to an application code) in both tchannel and grpc.
+Currently, unknown headers with the prefix `rpc-` are forwarded to the application code in both transports.
+- Any attempt to set request/response header with the prefix `rpc-` will result in an error in tchannel.
+Currently, the same behavior is applied only to the grpc transport, while tchannel allows setting such headers.
+
+As an intermediate step, the `reserved_headers_stripped` and `reserved_headers_error` metrics
+with `"component": "yarpc-header-migration"` constant tag and with `source` and `dest` variable tags
+will be emitted to help to identify the edges that are affected by the changes.
 
 ## [1.72.1] - 2024-03-14
 - tchannel: Renamed caller-procedure header from `$rpc$-caller-procedure` to `rpc-caller-procedure`.

docs/headers-handling.md

@@ -0,0 +1,86 @@
+# Headers handling
+
+Yarpc has unified API for getting and setting headers. Although implementations may wary 
+significantly from one transport to another.
+
+This document describes details of headers handling in Yarpc.
+
+# Existing behaviour
+
+## HTTP
+
+### Outbound - Request (writing via req.Headers.With)
+
+All application headers are prepended with an 'Rpc-Header' prefix.
+
+### Inbound - Request (Parsing)
+
+Predefined list of headers is read and stripped from the inbound request.
+
+Headers with prefix 'Rpc-Header-' will be forwarded to an application code (without prefix).
+
+Only headers explicitly specified in the config will be passed to the application code as is. If header name doesn't have 'x-' prefix, 'header %s does not begin with 'x-'' message is returned.
+
+### Inbound - Response (Writing)
+
+All application headers are prepended with an 'Rpc-Header' prefix.
+
+### Outbound - Response (Parsing)
+
+Headers with prefix 'Rpc-Header-' will be forwarded to an application code (without prefix).
+
+## TChannel
+
+### Outbound - Request (writing via req.Headers.With)
+
+Headers with any name may be added.
+
+### Inbound - Request (Parsing)
+
+Predefined list of headers (one header, actually) is read and stripped from the inbound request.
+
+All other headers are forwarded as is to an application code.
+
+### Inbound - Response (Writing)
+
+Attempting to add a header with a name listed as reserved leads to an error "cannot use reserved header key".
+
+### Outbound - Response (Parsing)
+
+Headers with the names listed as reserved are deleted. All other headers are forwarded to an application code as is.
+
+## GRPC
+
+### Outbound - Request (writing via req.Headers.With)
+
+Attempting to add headers with some of reserved names or with already set values lead to 'duplicate key' error.
+
+Attempting to add headers with 'rpc-' prefix leads to 'cannot use reserved header in application headers' error.
+
+### Inbound - Request (Parsing)
+
+Predefined list of headers is read and stripped from the inbound request.
+
+All other headers are forwarded as is to an application code.
+
+### Inbound - Response (Writing)
+
+Attempting to add headers with some of reserved names or already set values lead to 'duplicate key' error.
+
+Attempting to add headers with 'rpc-' prefix leads to 'cannot use reserved header in application headers' error.
+
+### Outbound - Response (Parsing)
+
+Headers with 'rpc-' prefix will be omitted from forwarding to an application code.
+
+# New behaviour
+
+## HTTP, TChannel, GRPC
+
+### Outbound - Request (writing via req.Headers.With) and Inbound - Response (Writing)
+
+Attempting to add a header with a 'prc-' or '$rpc$-' prefixes leads to an error "cannot use reserved header key".
+
+### Inbound - Request (Parsing) and Outbound - Response (Parsing)
+
+Unparsed headers with 'rpc-' or '$rpc$-' prefixes ignored, i.e. not forwarded to an application code.

internal/observability/public.go

@@ -0,0 +1,76 @@
+// Copyright (c) 2024 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package observability
+
+import (
+	"sync"
+
+	"go.uber.org/net/metrics"
+)
+
+var (
+	reservedHeaderStripped *metrics.CounterVector
+	reservedHeaderError    *metrics.CounterVector
+
+	registerHeaderMetricsOnce sync.Once
+)
+
+// IncReservedHeaderStripped increments the counter for reserved headers being stripped.
+func IncReservedHeaderStripped(m *metrics.Scope, source, dest string) {
+	registerHeaderMetrics(m)
+	incHeaderMetric(reservedHeaderStripped, source, dest)
+}
+
+// IncReservedHeaderError increments the counter for reserved headers led to error.
+func IncReservedHeaderError(m *metrics.Scope, source, dest string) {
+	registerHeaderMetrics(m)
+	incHeaderMetric(reservedHeaderError, source, dest)
+}
+
+func registerHeaderMetrics(m *metrics.Scope) {
+	if m == nil {
+		return
+	}
+
+	registerHeaderMetricsOnce.Do(func() {
+		reservedHeaderStripped, _ = m.CounterVector(metrics.Spec{
+			Name:      "reserved_headers_stripped",
+			Help:      "Total number of reserved headers being stripped.",
+			ConstTags: map[string]string{"component": "yarpc-header-migration"},
+			VarTags:   []string{"source", "dest"},
+		})
+
+		reservedHeaderError, _ = m.CounterVector(metrics.Spec{
+			Name:      "reserved_headers_error",
+			Help:      "Total number of reserved headers led to error.",
+			ConstTags: map[string]string{"component": "yarpc-header-migration"},
+			VarTags:   []string{"source", "dest"},
+		})
+	})
+}
+
+func incHeaderMetric(vector *metrics.CounterVector, source, dest string) {
+	if vector != nil {
+		if counter, err := vector.Get("source", source, "dest", dest); counter != nil && err == nil {
+			counter.Inc()
+		}
+	}
+}

internal/observability/public_test.go

@@ -0,0 +1,99 @@
+// Copyright (c) 2024 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package observability
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/net/metrics"
+)
+
+func TestReservedHeaderMetrics(t *testing.T) {
+	m := metrics.New()
+
+	t.Run("nil-scope", func(t *testing.T) {
+		IncReservedHeaderStripped(nil, "", "")
+		IncReservedHeaderError(nil, "", "")
+	})
+
+	t.Run("nil-counters", func(t *testing.T) {
+		// Counters registration called only once
+		registerHeaderMetrics(m.Scope())
+
+		var (
+			registeredStripped = reservedHeaderStripped
+			registeredError    = reservedHeaderError
+		)
+		t.Cleanup(func() {
+			reservedHeaderStripped = registeredStripped
+			reservedHeaderError = registeredError
+		})
+		reservedHeaderStripped = nil
+		reservedHeaderError = nil
+
+		IncReservedHeaderStripped(m.Scope(), "", "")
+		IncReservedHeaderError(m.Scope(), "", "")
+	})
+
+	t.Run("inc-header-metric", func(t *testing.T) {
+		IncReservedHeaderStripped(m.Scope(), "source", "dest")
+		IncReservedHeaderError(m.Scope(), "source", "dest")
+
+		IncReservedHeaderStripped(m.Scope(), "source", "dest")
+		IncReservedHeaderStripped(m.Scope(), "source", "dest-2")
+		IncReservedHeaderStripped(m.Scope(), "source-2", "dest-2")
+
+		s := m.Snapshot()
+
+		var (
+			strippedFound, errorFound bool
+		)
+		for _, c := range s.Counters {
+			if c.Name == "reserved_headers_stripped" {
+				strippedFound = true
+
+				if c.Tags["source"] == "source" && c.Tags["dest"] == "dest" {
+					assert.Equal(t, int64(2), c.Value)
+				} else if c.Tags["source"] == "source" && c.Tags["dest"] == "dest-2" {
+					assert.Equal(t, int64(1), c.Value)
+				} else if c.Tags["source"] == "source-2" && c.Tags["dest"] == "dest-2" {
+					assert.Equal(t, int64(1), c.Value)
+				} else {
+					t.Errorf("unexpected counter: %v", c)
+				}
+			} else if c.Name == "reserved_headers_error" {
+				errorFound = true
+
+				if c.Tags["source"] == "source" && c.Tags["dest"] == "dest" {
+					assert.Equal(t, int64(1), c.Value)
+				} else {
+					t.Errorf("unexpected counter: %v", c)
+				}
+			} else {
+				t.Errorf("unexpected counter: %v", c)
+			}
+		}
+
+		assert.True(t, strippedFound)
+		assert.True(t, errorFound)
+	})
+}

transport/grpc/handler.go

@@ -29,6 +29,7 @@ import (
 	"go.uber.org/yarpc/api/transport"
 	"go.uber.org/yarpc/internal/bufferpool"
 	"go.uber.org/yarpc/internal/grpcerrorcodes"
+	"go.uber.org/yarpc/internal/observability"
 	"go.uber.org/yarpc/yarpcerrors"
 	"go.uber.org/zap"
 	"golang.org/x/net/context"
@@ -88,7 +89,10 @@ func (h *handler) getBasicTransportRequest(ctx context.Context, streamMethod str
 	if md == nil || !ok {
 		return nil, yarpcerrors.Newf(yarpcerrors.CodeInternal, "cannot get metadata from ctx: %v", ctx)
 	}
-	transportRequest, err := metadataToTransportRequest(md)
+	transportRequest, reportHeader, err := metadataToInboundRequest(md)
+	if reportHeader {
+		observability.IncReservedHeaderStripped(h.i.t.options.meter, transportRequest.Caller, transportRequest.Service)
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -190,6 +194,10 @@ func (h *handler) handleUnary(
 	err := h.handleUnaryBeforeErrorConversion(ctx, transportRequest, responseWriter, start, handler)
 	err = handlerErrorToGRPCError(err, responseWriter)
 
+	if responseWriter.reportHeader {
+		observability.IncReservedHeaderError(h.i.t.options.meter, transportRequest.Caller, transportRequest.Service)
+	}
+
 	// Send the response attributes back and end the stream.
 	//
 	// Warning: SendMsg() holds onto these bytes after returning. Therefore, we