fix(kuma-cp): disable statsForAllMethods in grpc stats

UPGRADE.md

@@ -36,6 +36,13 @@ However, Zone Ingress Token is now deprecated and will be removed in the future.
 - By default, the minimum TLS version allowed on servers is TLSv1.2. If you require using TLS < 1.2 you can set `KUMA_GENERAL_TLS_MIN_VERSION`.
 - `KUMA_MONITORING_ASSIGNMENT_SERVER_GRPC_PORT` was removed after a long deprecation period use `KUMA_MONITORING_ASSIGNMENT_SERVER_PORT` instead.
 
+### gRPC metrics
+
+With this release, emitting separate statistics for every gRPC method is disabled.
+gRPC metrics from different methods are now aggregated under `envoy_cluster_grpc_request_message_count`.
+It will be re-enabled again in the future once Envoy with [`replace_dots_in_grpc_service_name`](https://www.envoyproxy.io/docs/envoy/latest/api-v3/extensions/filters/http/grpc_stats/v3/config.proto#envoy-v3-api-field-extensions-filters-http-grpc-stats-v3-filterconfig-stats-for-all-methods) feature is released.
+If you need to enable this setting, you can use ProxyTemplate to patch `envoy.filters.http.grpc_stats` http filter.
+
 ## Upgrade to `1.8.x`
 
 ### Kumactl

pkg/xds/envoy/listeners/v3/grpc_stats_configurer.go

@@ -16,9 +16,6 @@ var _ FilterChainConfigurer = &GrpcStatsConfigurer{}
 func (g *GrpcStatsConfigurer) Configure(filterChain *envoy_listener.FilterChain) error {
 	config := &envoy_grpc_stats.FilterConfig{
 		EmitFilterState: true,
-		PerMethodStatSpecifier: &envoy_grpc_stats.FilterConfig_StatsForAllMethods{
-			StatsForAllMethods: util_proto.Bool(true),
-		},
 	}
 	pbst, err := util_proto.MarshalAnyDeterministic(config)
 	if err != nil {

pkg/xds/envoy/listeners/v3/grpc_stats_configurer_test.go

@@ -39,7 +39,6 @@ var _ = Describe("gRPCStatsConfigurer", func() {
                   typedConfig:
                     '@type': type.googleapis.com/envoy.extensions.filters.http.grpc_stats.v3.FilterConfig
                     emitFilterState: true
-                    statsForAllMethods: true
                 - name: envoy.filters.http.router
                   typedConfig:
                     '@type': type.googleapis.com/envoy.extensions.filters.http.router.v3.Router

pkg/xds/generator/testdata/outbound-proxy/03.envoy.golden.yaml

@@ -454,7 +454,6 @@ resources:
             typedConfig:
               '@type': type.googleapis.com/envoy.extensions.filters.http.grpc_stats.v3.FilterConfig
               emitFilterState: true
-              statsForAllMethods: true
           - name: envoy.filters.http.router
             typedConfig:
               '@type': type.googleapis.com/envoy.extensions.filters.http.router.v3.Router

test/e2e_env/universal/grpc/grpc.go

@@ -0,0 +1,55 @@
+package grpc
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/kumahq/kuma/test/e2e_env/universal/env"
+	. "github.com/kumahq/kuma/test/framework"
+)
+
+func GRPC() {
+	meshName := "grpc"
+
+	BeforeAll(func() {
+		Expect(NewClusterSetup().
+			Install(MeshUniversal(meshName)).
+			Install(TestServerUniversal("test-server", meshName,
+				WithServiceName("test-server"),
+				WithProtocol("grpc"),
+				WithArgs([]string{"grpc", "server", "--port", "8080"}),
+				WithTransparentProxy(true),
+			)).
+			Install(TestServerUniversal("test-client", meshName,
+				WithServiceName("test-client"),
+				WithArgs([]string{"grpc", "client", "--address", "test-server.mesh:80"}),
+				WithTransparentProxy(true),
+			)).
+			Setup(env.Cluster)).To(Succeed())
+	})
+
+	E2EAfterAll(func() {
+		Expect(env.Cluster.DeleteMeshApps(meshName)).To(Succeed())
+		Expect(env.Cluster.DeleteMesh(meshName)).To(Succeed())
+	})
+
+	It("should emit stats from the server", func() {
+		Eventually(func(g Gomega) {
+			stdout, _, err := env.Cluster.Exec("", "", "test-server",
+				"curl", "-v", "--fail", "http://localhost:9901/stats?format=prometheus")
+			g.Expect(err).ToNot(HaveOccurred())
+			g.Expect(stdout).To(ContainSubstring(`envoy_cluster_grpc_request_message_count{envoy_cluster_name="localhost_8080"}`))
+			g.Expect(stdout).To(ContainSubstring(`envoy_cluster_grpc_response_message_count{envoy_cluster_name="localhost_8080"}`))
+		}, "30s", "1s").Should(Succeed())
+	})
+
+	It("should emit stats from the client", func() {
+		Eventually(func(g Gomega) {
+			stdout, _, err := env.Cluster.Exec("", "", "test-client",
+				"curl", "-v", "--fail", "http://localhost:9901/stats?format=prometheus")
+			g.Expect(err).ToNot(HaveOccurred())
+			g.Expect(stdout).To(ContainSubstring(`envoy_cluster_grpc_request_message_count{envoy_cluster_name="test-server"}`))
+			g.Expect(stdout).To(ContainSubstring(`envoy_cluster_grpc_response_message_count{envoy_cluster_name="test-server"}`))
+		}, "30s", "1s").Should(Succeed())
+	})
+}

test/e2e_env/universal/universal_suite_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/kumahq/kuma/test/e2e_env/universal/env"
 	"github.com/kumahq/kuma/test/e2e_env/universal/externalservices"
 	"github.com/kumahq/kuma/test/e2e_env/universal/gateway"
+	"github.com/kumahq/kuma/test/e2e_env/universal/grpc"
 	"github.com/kumahq/kuma/test/e2e_env/universal/healthcheck"
 	"github.com/kumahq/kuma/test/e2e_env/universal/inspect"
 	"github.com/kumahq/kuma/test/e2e_env/universal/matching"
@@ -114,3 +115,4 @@ var _ = Describe("Zone Egress", zoneegress.ExternalServices, Ordered)
 var _ = Describe("Virtual Outbound", virtualoutbound.VirtualOutbound, Ordered)
 var _ = Describe("Transparent Proxy", transparentproxy.TransparentProxy, Ordered)
 var _ = Describe("Mesh Traffic Permission", meshtrafficpermission.MeshTrafficPermissionUniversal, Ordered)
+var _ = Describe("GRPC", grpc.GRPC, Ordered)

test/framework/setup.go

@@ -557,7 +557,9 @@ func TestServerUniversal(name string, mesh string, opt ...AppDeploymentOption) I
 			serviceAddress = fmt.Sprintf(`    serviceAddress: %s`, opts.serviceAddress)
 		}
 
-		args = append(args, "--port", "8080")
+		if len(args) < 2 || args[1] != "grpc" { // grpc client does not have port
+			args = append(args, "--port", "8080")
+		}
 		appYaml := fmt.Sprintf(`
 type: Dataplane
 mesh: %s

test/server/cmd/grpc_client.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -28,54 +29,57 @@ func newGRPCClientCmd() *cobra.Command {
 # sends "Request #${n_of_streams}.${n_of_requests}" every 2 seconds.
 test-server grpc client address="localhost:8080"`,
 		RunE: func(cmd *cobra.Command, _ []string) error {
-			conn, err := grpc.Dial(args.address,
-				grpc.WithTransportCredentials(insecure.NewCredentials()),
-				grpc.WithKeepaliveParams(keepalive.ClientParameters{
-					Time:    10 * time.Second,
-					Timeout: 10 * time.Second,
-				}),
-			)
-			if err != nil {
-				grpcClientLog.Error(err, "dial failed")
-				return err
-			}
-			defer conn.Close()
-
-			c := api.NewGreeterClient(conn)
-
 			streamCounter := 0
 			for {
-				clientStream, err := c.SayHellos(context.Background())
-				if err != nil {
-					grpcClientLog.Error(err, "SayHellos failed")
-					return err
+				streamCounter++
+				if err := startSendingRequests(args.address, streamCounter); err != nil {
+					grpcClientLog.Error(err, "sending requests failed")
 				}
+				time.Sleep(1 * time.Second)
+			}
+		},
+	}
+	cmd.PersistentFlags().StringVar(&args.address, "address", "localhost:8080", "server address")
+	return cmd
+}
 
-				requestCounter := 0
-				for {
-					err := clientStream.Send(&api.HelloRequest{
-						Name: fmt.Sprintf("Request #%d.%d", streamCounter, requestCounter),
-					})
-					if err != nil {
-						grpcClientLog.Error(err, "Send failed")
-						break
-					}
+func startSendingRequests(address string, streamCounter int) error {
+	conn, err := grpc.Dial(address,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithKeepaliveParams(keepalive.ClientParameters{
+			Time:    10 * time.Second,
+			Timeout: 10 * time.Second,
+		}),
+	)
+	if err != nil {
+		return errors.Wrap(err, "dial failed")
+	}
+	defer conn.Close()
 
-					resp, err := clientStream.Recv()
-					if err != nil {
-						grpcClientLog.Error(err, "Recv failed")
-						break
-					}
-					grpcClientLog.Info("received response", "msg", resp.GetMessage())
+	c := api.NewGreeterClient(conn)
+	for {
+		clientStream, err := c.SayHellos(context.Background())
+		if err != nil {
+			return errors.Wrap(err, "SayHellos failed")
+		}
 
-					time.Sleep(2 * time.Second)
-					requestCounter++
-				}
+		requestCounter := 0
+		for {
+			err := clientStream.Send(&api.HelloRequest{
+				Name: fmt.Sprintf("Request #%d.%d", streamCounter, requestCounter),
+			})
+			if err != nil {
+				return errors.Wrap(err, "Send failed")
+			}
 
-				streamCounter++
+			resp, err := clientStream.Recv()
+			if err != nil {
+				return errors.Wrap(err, "Recv failed")
 			}
-		},
+			grpcClientLog.Info("received response", "msg", resp.GetMessage())
+
+			time.Sleep(2 * time.Second)
+			requestCounter++
+		}
 	}
-	cmd.PersistentFlags().StringVar(&args.address, "address", "localhost:8080", "server address")
-	return cmd
 }