kuma-cp: add support for %KUMA_*% placeholders inside Envoy access lo…

…g format

pkg/envoy/accesslog/commands.go

@@ -2,6 +2,7 @@ package accesslog
 
 import (
 	"fmt"
+	"strings"
 )
 
 // List of supported command operators.
@@ -61,6 +62,12 @@ const (
 	CMD_DOWNSTREAM_PEER_CERT_V_START    = "DOWNSTREAM_PEER_CERT_V_START"
 	CMD_DOWNSTREAM_PEER_CERT_V_END      = "DOWNSTREAM_PEER_CERT_V_END"
 	CMD_HOSTNAME                        = "HOSTNAME"
+
+	// Commands unique to Kuma.
+
+	CMD_KUMA_SOURCE_ADDRESS      = "KUMA_SOURCE_ADDRESS"
+	CMD_KUMA_SOURCE_SERVICE      = "KUMA_SOURCE_SERVICE"
+	CMD_KUMA_DESTINATION_SERVICE = "KUMA_DESTINATION_SERVICE"
 )
 
 // CommandOperatorDescriptor represents a descriptor of an Envoy access log command operator.
@@ -156,7 +163,21 @@ func (o CommandOperatorDescriptor) String() string {
 		return "%DOWNSTREAM_PEER_CERT_V_END%"
 	case CMD_HOSTNAME:
 		return "%HOSTNAME%"
+	case CMD_KUMA_SOURCE_ADDRESS:
+		return "%KUMA_SOURCE_ADDRESS%"
+	case CMD_KUMA_SOURCE_SERVICE:
+		return "%KUMA_SOURCE_SERVICE%"
+	case CMD_KUMA_DESTINATION_SERVICE:
+		return "%KUMA_DESTINATION_SERVICE%"
 	default:
 		return fmt.Sprintf("%%%s%%", string(o))
 	}
 }
+
+// IsPlaceholder returns true if this command is a placeholder
+// that must be resolved before configuring Envoy with that format string.
+// E.g., %KUMA_SOURCE_ADDRESS%, %KUMA_SOURCE_SERVICE% and %KUMA_DESTINATION_SERVICE%
+// are examples of such placeholders.
+func (o CommandOperatorDescriptor) IsPlaceholder() bool {
+	return strings.HasPrefix(string(o), "KUMA_")
+}

pkg/envoy/accesslog/format.go

@@ -64,6 +64,27 @@ func (f *AccessLogFormat) ConfigureTcpLog(config *accesslog_config.TcpGrpcAccess
 	return nil
 }
 
+func (f *AccessLogFormat) Interpolate(context InterpolationContext) (*AccessLogFormat, error) {
+	newFragments := make([]AccessLogFragment, len(f.Fragments))
+	interpolated := false
+	for i, fragment := range f.Fragments {
+		if interpolator, ok := fragment.(AccessLogFragmentInterpolator); ok {
+			newFragment, err := interpolator.Interpolate(context)
+			if err != nil {
+				return nil, err
+			}
+			newFragments[i] = newFragment
+			interpolated = interpolated || newFragment != fragment
+		} else {
+			newFragments[i] = fragment
+		}
+	}
+	if interpolated {
+		return &AccessLogFormat{Fragments: newFragments}, nil
+	}
+	return f, nil
+}
+
 // String returns the canonical representation of this format string.
 func (f *AccessLogFormat) String() string {
 	fragments := make([]string, len(f.Fragments))

pkg/envoy/accesslog/format_parser.go

@@ -25,16 +25,16 @@ var parser = formatParser{}
 //   2. To adjust configuration of `envoy.http_grpc_access_log` and `envoy.tcp_grpc_access_log`
 //      according to the format string, e.g. to capture additional HTTP headers
 //   3. To format a given HTTP or TCP log entry according to the format string
-//   4. (not implemented yet) To bind `%KUMA_*%` placeholders to concrete context-dependent values
-//   5. (not implemented yet) To render back into the format string, e.g.
+//   4. To bind `%KUMA_*%` placeholders to concrete context-dependent values
+//   5. To render back into the format string, e.g.
 //      after `%KUMA_*%` placeholders have been bound to concrete context-dependent values
-func ParseFormat(format string) (AccessLogFragment, error) {
+func ParseFormat(format string) (*AccessLogFormat, error) {
 	return parser.Parse(format)
 }
 
 type formatParser struct{}
 
-func (p formatParser) Parse(format string) (_ AccessLogFragment, err error) {
+func (p formatParser) Parse(format string) (_ *AccessLogFormat, err error) {
 	defer func() {
 		if err != nil {
 			err = errors.Wrap(err, "format string is not valid")
@@ -74,7 +74,7 @@ func (p formatParser) Parse(format string) (_ AccessLogFragment, err error) {
 		fragments = append(fragments, TextSpan(format[textLiteralStart:]))
 	}
 
-	return &AccessLogFormat{fragments}, nil
+	return &AccessLogFormat{Fragments: fragments}, nil
 }
 
 func (p formatParser) splitMatch(match []string) (token string, command string, args string, limit string, err error) {
@@ -127,6 +127,9 @@ func (p formatParser) parseCommandOperator(token, command, args, limit string) (
 		if err != nil {
 			return nil, err
 		}
+		if CommandOperatorDescriptor(field).IsPlaceholder() {
+			return Placeholder(field), nil
+		}
 		return FieldOperator(field), nil
 	}
 }

pkg/envoy/accesslog/format_parser_test.go

@@ -540,6 +540,21 @@ var _ = Describe("ParseFormat()", func() {
 				expectedHTTP: `UNSUPPORTED_COMMAND(%HOSTNAME%)`,
 				expectedTCP:  `UNSUPPORTED_COMMAND(%HOSTNAME%)`,
 			}),
+			Entry("%KUMA_SOURCE_ADDRESS%", testCase{
+				format:       `%KUMA_SOURCE_ADDRESS%`,
+				expectedHTTP: `%KUMA_SOURCE_ADDRESS%`, // placeholder must be rendered "as is"
+				expectedTCP:  `%KUMA_SOURCE_ADDRESS%`,
+			}),
+			Entry("%KUMA_SOURCE_SERVICE%", testCase{
+				format:       `%KUMA_SOURCE_SERVICE%`,
+				expectedHTTP: `%KUMA_SOURCE_SERVICE%`, // placeholder must be rendered "as is"
+				expectedTCP:  `%KUMA_SOURCE_SERVICE%`,
+			}),
+			Entry("%KUMA_DESTINATION_SERVICE%", testCase{
+				format:       `%KUMA_DESTINATION_SERVICE%`,
+				expectedHTTP: `%KUMA_DESTINATION_SERVICE%`, // placeholder must be rendered "as is"
+				expectedTCP:  `%KUMA_DESTINATION_SERVICE%`,
+			}),
 			Entry("composite", testCase{
 				format:       `[%START_TIME%] "%REQ(:METHOD)% %REQ(X-ENVOY-ORIGINAL-PATH?:PATH)% %PROTOCOL%" %RESPONSE_CODE% %RESPONSE_FLAGS% %BYTES_RECEIVED% %BYTES_SENT% %DURATION% %RESP(X-ENVOY-UPSTREAM-SERVICE-TIME)% "%REQ(X-FORWARDED-FOR)%" "%REQ(USER-AGENT)%" "%REQ(X-REQUEST-ID)%" "%REQ(:AUTHORITY)%"`,
 				expectedHTTP: `[2020-02-18T21:52:17.987Z] "- /api HTTP/1.1" 200 UF,URX 234 567 123 - "-" "-" "-" "backend.internal:8080"`,
@@ -794,6 +809,8 @@ UF,URX
 "%TRAILER(GRPC-STATUS):1%" "%TRAILER(grpc-message):2%" "%TRAILER(grpc-status):3%"
 "%DYNAMIC_METADATA(com.test.my_filter:test_object:inner_key_1):1%" "%DYNAMIC_METADATA(com.test.my_filter:test_object:inner_key_2):2%" "%DYNAMIC_METADATA(com.test.my_filter:test_object:inner_key_1):3%"
 "%FILTER_STATE(filter.state.key1):1%" "%FILTER_STATE(filter.state.key2):2%" "%FILTER_STATE(filter.state.key1):3%"
+%BYTES_SENT%
+%KUMA_SOURCE_SERVICE%
 `,
 				expectedHTTP: &accesslog_config.HttpGrpcAccessLogConfig{
 					CommonConfig: &accesslog_config.CommonGrpcAccessLogConfig{
@@ -829,7 +846,6 @@ UF,URX
 				actual := format.String()
 				// then
 				Expect(actual).To(Equal(given.expected))
-
 			},
 			Entry("composite", testCase{
 				format:   `[%START_TIME%] "%REQ(:METHOD)% %REQ(X-ENVOY-ORIGINAL-PATH?:PATH)% %PROTOCOL%" %RESPONSE_CODE% %RESPONSE_FLAGS% %BYTES_RECEIVED% %BYTES_SENT% %DURATION% %RESP(X-ENVOY-UPSTREAM-SERVICE-TIME)% "%REQ(X-FORWARDED-FOR)%" "%REQ(USER-AGENT)%" "%REQ(X-REQUEST-ID)%" "%REQ(:AUTHORITY)%"`,
@@ -881,6 +897,9 @@ UF,URX
 %DOWNSTREAM_PEER_CERT_V_START%
 %DOWNSTREAM_PEER_CERT_V_END%
 %HOSTNAME%
+%KUMA_SOURCE_ADDRESS%
+%KUMA_SOURCE_SERVICE%
+%KUMA_DESTINATION_SERVICE%
 `,
 				expected: `
 %START_TIME%
@@ -927,6 +946,82 @@ UF,URX
 %DOWNSTREAM_PEER_CERT_V_START%
 %DOWNSTREAM_PEER_CERT_V_END%
 %HOSTNAME%
+%KUMA_SOURCE_ADDRESS%
+%KUMA_SOURCE_SERVICE%
+%KUMA_DESTINATION_SERVICE%
+`,
+			}),
+		)
+	})
+
+	Describe("support Interpolate()", func() {
+		type testCase struct {
+			format   string
+			context  map[string]string
+			expected string
+		}
+
+		DescribeTable("should bind to a given context",
+			func(given testCase) {
+				// when
+				format, err := ParseFormat(given.format)
+				// then
+				Expect(err).ToNot(HaveOccurred())
+
+				// when
+				interpolatedFormat, err := format.Interpolate(InterpolationVariables(given.context))
+				// then
+				Expect(err).ToNot(HaveOccurred())
+
+				// when
+				actual := interpolatedFormat.String()
+				// then
+				Expect(actual).To(Equal(given.expected))
+			},
+			Entry("multi-line w/ empty context", testCase{
+				format: `
+%START_TIME%
+%KUMA_SOURCE_ADDRESS%
+%BYTES_RECEIVED%
+%KUMA_SOURCE_SERVICE%
+%BYTES_SENT%
+%KUMA_DESTINATION_SERVICE%
+%PROTOCOL%
+`,
+				context: nil,
+				expected: `
+%START_TIME%
+
+%BYTES_RECEIVED%
+
+%BYTES_SENT%
+
+%PROTOCOL%
+`,
+			}),
+			Entry("multi-line w/ full context", testCase{
+				format: `
+%START_TIME%
+%KUMA_SOURCE_ADDRESS%
+%BYTES_RECEIVED%
+%KUMA_SOURCE_SERVICE%
+%BYTES_SENT%
+%KUMA_DESTINATION_SERVICE%
+%PROTOCOL%
+`,
+				context: map[string]string{
+					"KUMA_SOURCE_ADDRESS":      "10.0.0.3:0",
+					"KUMA_SOURCE_SERVICE":      "web",
+					"KUMA_DESTINATION_SERVICE": "backend",
+				},
+				expected: `
+%START_TIME%
+10.0.0.3:0
+%BYTES_RECEIVED%
+web
+%BYTES_SENT%
+backend
+%PROTOCOL%
 `,
 			}),
 		)

pkg/envoy/accesslog/interfaces.go

@@ -41,3 +41,26 @@ type HttpLogConfigurer interface {
 type TcpLogConfigurer interface {
 	ConfigureTcpLog(config *accesslog_config.TcpGrpcAccessLogConfig) error
 }
+
+// InterpolationContext represents a context of Interpolate() operation.
+type InterpolationContext interface {
+	// Get returns a variable value in this context.
+	Get(variable string) string
+}
+
+// AccessLogFragmentInterpolator interpolates placeholders
+// added to an access log format string.
+// E.g. %KUMA_SOURCE_SERVICE%, %KUMA_DESTINATION_SERVICE% and %KUMA_SOURCE_ADDRESS%
+// are examples of such placeholders.
+type AccessLogFragmentInterpolator interface {
+	// Interpolate returns an access log fragment with all placeholders resolved.
+	Interpolate(context InterpolationContext) (AccessLogFragment, error)
+}
+
+// InterpolationVariables represents a context of Interpolate() operation
+// as a map of variables.
+type InterpolationVariables map[string]string
+
+func (m InterpolationVariables) Get(variable string) string {
+	return m[variable]
+}

pkg/envoy/accesslog/placeholder_fragment.go

@@ -0,0 +1,42 @@
+package accesslog
+
+import (
+	accesslog_config "github.com/envoyproxy/go-control-plane/envoy/config/accesslog/v2"
+	accesslog_data "github.com/envoyproxy/go-control-plane/envoy/data/accesslog/v2"
+)
+
+// Placeholder represents a placeholder added to an access log format string
+// that must be resolved before configuring Envoy with that format string.
+//
+// E.g. %KUMA_SOURCE_SERVICE%, %KUMA_DESTINATION_SERVICE% and %KUMA_SOURCE_ADDRESS%
+// are examples of such placeholders.
+type Placeholder string
+
+func (f Placeholder) FormatHttpLogEntry(entry *accesslog_data.HTTPAccessLogEntry) (string, error) {
+	return f.String(), nil
+}
+
+func (f Placeholder) FormatTcpLogEntry(entry *accesslog_data.TCPAccessLogEntry) (string, error) {
+	return f.String(), nil
+}
+
+func (f Placeholder) ConfigureHttpLog(config *accesslog_config.HttpGrpcAccessLogConfig) error {
+	// has no effect on HttpGrpcAccessLogConfig
+	return nil
+}
+
+func (f Placeholder) ConfigureTcpLog(config *accesslog_config.TcpGrpcAccessLogConfig) error {
+	// has no effect on TcpGrpcAccessLogConfig
+	return nil
+}
+
+// String returns the canonical representation of this command operator.
+func (f Placeholder) String() string {
+	return CommandOperatorDescriptor(f).String()
+}
+
+// Interpolate returns an access log fragment with all placeholders resolved.
+func (f Placeholder) Interpolate(context InterpolationContext) (AccessLogFragment, error) {
+	value := context.Get(string(f))
+	return TextSpan(value), nil // turn placeholder into a text literal
+}