[Feature] [Networking] ArangoRotue WebSocket Support (#1759)

CHANGELOG.md

@@ -15,6 +15,7 @@
 - (Feature) (Platform) Storage V1Alpha1
 - (Feature) StorageV2 Integration Service Implementation
 - (Feature) (Platform) Storage V1Alpha1 RC
+- (Feature) (Networking) ArangoRotue WebSocket Support
 
 ## [1.2.43](https://github.com/arangodb/kube-arangodb/tree/1.2.43) (2024-10-14)
 - (Feature) ArangoRoute CRD

docs/api/ArangoRoute.V1Alpha1.md

@@ -145,6 +145,25 @@ Insecure allows Insecure traffic
 
 ***
 
+### .spec.options.upgrade\[int\].enabled
+
+Type: `boolean` <sup>[\[ref\]](https://github.com/arangodb/kube-arangodb/blob/1.2.43/pkg/apis/networking/v1alpha1/route_spec_options_upgrade.go#L37)</sup>
+
+Enabled defines if upgrade option is enabled
+
+***
+
+### .spec.options.upgrade\[int\].type
+
+Type: `string` <sup>[\[ref\]](https://github.com/arangodb/kube-arangodb/blob/1.2.43/pkg/apis/networking/v1alpha1/route_spec_options_upgrade.go#L34)</sup>
+
+Type defines type of the Upgrade
+
+Possible Values: 
+* `"websocket"` (default) - HTTP WebSocket Upgrade type
+
+***
+
 ### .spec.route.path
 
 Type: `string` <sup>[\[ref\]](https://github.com/arangodb/kube-arangodb/blob/1.2.43/pkg/apis/networking/v1alpha1/route_spec_route.go#L29)</sup>

pkg/apis/networking/v1alpha1/route_spec.go

@@ -31,6 +31,9 @@ type ArangoRouteSpec struct {
 
 	// Route defines the route spec
 	Route *ArangoRouteSpecRoute `json:"route,omitempty"`
+
+	// Options defines connection upgrade options
+	Options *ArangoRouteSpecOptions `json:"options,omitempty"`
 }
 
 func (s *ArangoRouteSpec) GetDeployment() string {
@@ -65,6 +68,7 @@ func (s *ArangoRouteSpec) Validate() error {
 		shared.PrefixResourceErrors("deployment", shared.ValidateResourceNamePointer(s.Deployment)),
 		shared.ValidateRequiredInterfacePath("destination", s.Destination),
 		shared.ValidateOptionalInterfacePath("route", s.Route),
+		shared.ValidateOptionalInterfacePath("options", s.Options),
 	)); err != nil {
 		return err
 	}

pkg/apis/networking/v1alpha1/route_spec_options.go

@@ -0,0 +1,42 @@
+//
+// DISCLAIMER
+//
+// Copyright 2024 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package v1alpha1
+
+import shared "github.com/arangodb/kube-arangodb/pkg/apis/shared"
+
+type ArangoRouteSpecOptions struct {
+	// Upgrade keeps the connection upgrade options
+	Upgrade ArangoRouteSpecOptionsUpgrade `json:"upgrade,omitempty"`
+}
+
+func (a *ArangoRouteSpecOptions) Validate() error {
+	if a == nil {
+		a = &ArangoRouteSpecOptions{}
+	}
+
+	if err := shared.WithErrors(
+		shared.ValidateOptionalInterfacePath("upgrade", a.Upgrade),
+	); err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/apis/networking/v1alpha1/route_spec_options_upgrade.go

@@ -0,0 +1,48 @@
+//
+// DISCLAIMER
+//
+// Copyright 2024 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package v1alpha1
+
+import shared "github.com/arangodb/kube-arangodb/pkg/apis/shared"
+
+type ArangoRouteSpecOptionsUpgrade []ArangoRouteSpecOptionUpgrade
+
+func (a ArangoRouteSpecOptionsUpgrade) Validate() error {
+	return shared.ValidateInterfaceList(a)
+}
+
+type ArangoRouteSpecOptionUpgrade struct {
+	// Type defines type of the Upgrade
+	// +doc/enum: websocket|HTTP WebSocket Upgrade type
+	Type ArangoRouteUpgradeOptionType `json:"type"`
+
+	// Enabled defines if upgrade option is enabled
+	Enabled *bool `json:"enabled,omitempty"`
+}
+
+func (a ArangoRouteSpecOptionUpgrade) Validate() error {
+	if err := shared.WithErrors(
+		shared.ValidateRequiredInterfacePath("type", a.Type),
+	); err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/apis/networking/v1alpha1/route_upgrade_type.go

@@ -0,0 +1,38 @@
+//
+// DISCLAIMER
+//
+// Copyright 2024 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package v1alpha1
+
+import "github.com/arangodb/kube-arangodb/pkg/util/errors"
+
+type ArangoRouteUpgradeOptionType string
+
+const (
+	ArangoRouteUpgradeOptionWebsocket ArangoRouteUpgradeOptionType = "websocket"
+)
+
+func (a ArangoRouteUpgradeOptionType) Validate() error {
+	switch a {
+	case ArangoRouteUpgradeOptionWebsocket:
+		return nil
+	default:
+		return errors.Errorf("Invalid UpgradeOptionType: %s", a)
+	}
+}

pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go

@@ -111,6 +111,11 @@ func (in *ArangoRouteSpec) DeepCopyInto(out *ArangoRouteSpec) {
 		*out = new(ArangoRouteSpecRoute)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Options != nil {
+		in, out := &in.Options, &out.Options
+		*out = new(ArangoRouteSpecOptions)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -269,6 +274,63 @@ func (in *ArangoRouteSpecDestinationTLS) DeepCopy() *ArangoRouteSpecDestinationT
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ArangoRouteSpecOptionUpgrade) DeepCopyInto(out *ArangoRouteSpecOptionUpgrade) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ArangoRouteSpecOptionUpgrade.
+func (in *ArangoRouteSpecOptionUpgrade) DeepCopy() *ArangoRouteSpecOptionUpgrade {
+	if in == nil {
+		return nil
+	}
+	out := new(ArangoRouteSpecOptionUpgrade)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ArangoRouteSpecOptions) DeepCopyInto(out *ArangoRouteSpecOptions) {
+	*out = *in
+	if in.Upgrade != nil {
+		in, out := &in.Upgrade, &out.Upgrade
+		*out = make(ArangoRouteSpecOptionsUpgrade, len(*in))
+		copy(*out, *in)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ArangoRouteSpecOptions.
+func (in *ArangoRouteSpecOptions) DeepCopy() *ArangoRouteSpecOptions {
+	if in == nil {
+		return nil
+	}
+	out := new(ArangoRouteSpecOptions)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in ArangoRouteSpecOptionsUpgrade) DeepCopyInto(out *ArangoRouteSpecOptionsUpgrade) {
+	{
+		in := &in
+		*out = make(ArangoRouteSpecOptionsUpgrade, len(*in))
+		copy(*out, *in)
+		return
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ArangoRouteSpecOptionsUpgrade.
+func (in ArangoRouteSpecOptionsUpgrade) DeepCopy() ArangoRouteSpecOptionsUpgrade {
+	if in == nil {
+		return nil
+	}
+	out := new(ArangoRouteSpecOptionsUpgrade)
+	in.DeepCopyInto(out)
+	return *out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ArangoRouteSpecRoute) DeepCopyInto(out *ArangoRouteSpecRoute) {
 	*out = *in

pkg/apis/shared/validate.go

@@ -181,6 +181,17 @@ func ValidateRequiredInterfacePath[T ValidateInterface](path string, in T) error
 	return PrefixResourceErrors(path, ValidateRequiredInterface(in))
 }
 
+// ValidateInterfaceList Validates object if is not nil with path
+func ValidateInterfaceList[T ValidateInterface](in []T) error {
+	errors := make([]error, len(in))
+
+	for id := range in {
+		errors[id] = PrefixResourceError(fmt.Sprintf("[%d]", id), in[id].Validate())
+	}
+
+	return WithErrors(errors...)
+}
+
 // ValidateList validates all elements on the list
 func ValidateList[T any](in []T, validator func(T) error, checks ...func(in []T) error) error {
 	errors := make([]error, len(in)+len(checks))

pkg/crd/crds/networking-route.schema.generated.yaml

@@ -80,6 +80,24 @@ v1alpha1:
                     type: boolean
                 type: object
             type: object
+          options:
+            description: Options defines connection upgrade options
+            properties:
+              upgrade:
+                description: Upgrade keeps the connection upgrade options
+                items:
+                  properties:
+                    enabled:
+                      description: Enabled defines if upgrade option is enabled
+                      type: boolean
+                    type:
+                      description: Type defines type of the Upgrade
+                      enum:
+                        - websocket
+                      type: string
+                  type: object
+                type: array
+            type: object
           route:
             description: Route defines the route spec
             properties:

pkg/deployment/resources/gateway/gateway_config_destination.go

@@ -67,6 +67,8 @@ type ConfigDestination struct {
 	Path *string `json:"path,omitempty"`
 
 	AuthExtension *ConfigAuthZExtension `json:"authExtension,omitempty"`
+
+	UpgradeConfigs ConfigDestinationsUpgrade `json:"upgradeConfigs,omitempty"`
 }
 
 func (c *ConfigDestination) Validate() error {
@@ -78,6 +80,7 @@ func (c *ConfigDestination) Validate() error {
 		shared.PrefixResourceError("type", c.Type.Validate()),
 		shared.PrefixResourceError("path", shared.ValidateAPIPath(c.GetPath())),
 		shared.PrefixResourceError("authExtension", c.AuthExtension.Validate()),
+		shared.PrefixResourceError("upgradeConfigs", c.UpgradeConfigs.Validate()),
 	)
 }
 
@@ -111,6 +114,7 @@ func (c *ConfigDestination) RenderRoute(name, prefix string) (*routeAPI.Route, e
 				ClusterSpecifier: &routeAPI.RouteAction_Cluster{
 					Cluster: name,
 				},
+				UpgradeConfigs: c.getUpgradeConfigs().render(),
 				PrefixRewrite:  c.GetPath(),
 			},
 		},
@@ -118,6 +122,14 @@ func (c *ConfigDestination) RenderRoute(name, prefix string) (*routeAPI.Route, e
 	}, nil
 }
 
+func (c *ConfigDestination) getUpgradeConfigs() ConfigDestinationsUpgrade {
+	if c == nil {
+		return nil
+	}
+
+	return c.UpgradeConfigs
+}
+
 func (c *ConfigDestination) RenderCluster(name string) (*clusterAPI.Cluster, error) {
 	hpo, err := anypb.New(&upstreamHttpApi.HttpProtocolOptions{
 		UpstreamProtocolOptions: &upstreamHttpApi.HttpProtocolOptions_ExplicitHttpConfig_{

pkg/deployment/resources/gateway/gateway_config_destination_upgrade.go

@@ -0,0 +1,72 @@
+//
+// DISCLAIMER
+//
+// Copyright 2024 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package gateway
+
+import (
+	routeAPI "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+
+	shared "github.com/arangodb/kube-arangodb/pkg/apis/shared"
+	"github.com/arangodb/kube-arangodb/pkg/util"
+	"github.com/arangodb/kube-arangodb/pkg/util/errors"
+)
+
+type ConfigDestinationsUpgrade []ConfigDestinationUpgrade
+
+func (c ConfigDestinationsUpgrade) render() []*routeAPI.RouteAction_UpgradeConfig {
+	if len(c) == 0 {
+		return nil
+	}
+
+	var r = make([]*routeAPI.RouteAction_UpgradeConfig, len(c))
+
+	for id := range c {
+		r[id] = c[id].render()
+	}
+
+	return r
+}
+
+func (c ConfigDestinationsUpgrade) Validate() error {
+	return shared.ValidateInterfaceList(c)
+}
+
+type ConfigDestinationUpgrade struct {
+	Type string `json:"type"`
+
+	Enabled *bool `json:"enabled,omitempty"`
+}
+
+func (c ConfigDestinationUpgrade) render() *routeAPI.RouteAction_UpgradeConfig {
+	return &routeAPI.RouteAction_UpgradeConfig{
+		UpgradeType: c.Type,
+		Enabled:     wrapperspb.Bool(util.OptionalType(c.Enabled, true)),
+	}
+}
+
+func (c ConfigDestinationUpgrade) Validate() error {
+	switch c.Type {
+	case "websocket":
+		return nil
+	default:
+		return shared.PrefixResourceError("type", errors.Errorf("Unknown type: %s", c.Type))
+	}
+}

pkg/deployment/resources/gateway/gateway_config_test.go

@@ -25,12 +25,16 @@ import (
 	"testing"
 
 	bootstrapAPI "github.com/envoyproxy/go-control-plane/envoy/config/bootstrap/v3"
+	httpConnectionManagerAPI "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3"
 	"github.com/stretchr/testify/require"
 
 	"github.com/arangodb/kube-arangodb/pkg/util"
+	"github.com/arangodb/kube-arangodb/pkg/util/tests/tgrpc"
 )
 
 func renderAndPrintGatewayConfig(t *testing.T, cfg Config, validates ...func(t *testing.T, b *bootstrapAPI.Bootstrap)) {
+	require.NoError(t, cfg.Validate())
+
 	data, checksum, obj, err := cfg.RenderYAML()
 	require.NoError(t, err)
 
@@ -75,6 +79,136 @@ func Test_GatewayConfig(t *testing.T) {
 			require.EqualValues(t, 12345, b.StaticResources.Clusters[0].LoadAssignment.Endpoints[0].LbEndpoints[0].GetEndpoint().Address.GetSocketAddress().GetPortValue())
 		})
 	})
+	t.Run("Without WebSocket", func(t *testing.T) {
+		renderAndPrintGatewayConfig(t, Config{
+			DefaultDestination: ConfigDestination{
+				Targets: []ConfigDestinationTarget{
+					{
+						Host: "127.0.0.1",
+						Port: 12345,
+					},
+				},
+			},
+		}, func(t *testing.T, b *bootstrapAPI.Bootstrap) {
+			require.NotNil(t, b)
+			require.NotNil(t, b.StaticResources)
+			require.NotNil(t, b.StaticResources.Listeners)
+			require.Len(t, b.StaticResources.Listeners, 1)
+			require.NotNil(t, b.StaticResources.Listeners[0])
+			require.NotNil(t, b.StaticResources.Listeners[0].DefaultFilterChain)
+			require.NotNil(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters)
+			require.Len(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters, 1)
+			require.NotNil(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters[0])
+			var o httpConnectionManagerAPI.HttpConnectionManager
+			tgrpc.GRPCAnyCastAs(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters[0].GetTypedConfig(), &o)
+			rc := o.GetRouteConfig()
+			require.NotNil(t, rc)
+			require.NotNil(t, rc.VirtualHosts)
+			require.Len(t, rc.VirtualHosts, 1)
+			require.NotNil(t, rc.VirtualHosts[0])
+			require.Len(t, rc.VirtualHosts[0].Routes, 1)
+			require.NotNil(t, rc.VirtualHosts[0].Routes[0])
+			r := rc.VirtualHosts[0].Routes[0].GetRoute()
+			require.NotNil(t, r)
+			require.Len(t, r.UpgradeConfigs, 0)
+		})
+	})
+
+	t.Run("With WebSocket", func(t *testing.T) {
+		renderAndPrintGatewayConfig(t, Config{
+			DefaultDestination: ConfigDestination{
+				Targets: []ConfigDestinationTarget{
+					{
+						Host: "127.0.0.1",
+						Port: 12345,
+					},
+				},
+				UpgradeConfigs: ConfigDestinationsUpgrade{
+					{
+						Type: "websocket",
+					},
+				},
+			},
+		}, func(t *testing.T, b *bootstrapAPI.Bootstrap) {
+			require.NotNil(t, b)
+			require.NotNil(t, b.StaticResources)
+			require.NotNil(t, b.StaticResources.Listeners)
+			require.Len(t, b.StaticResources.Listeners, 1)
+			require.NotNil(t, b.StaticResources.Listeners[0])
+			require.NotNil(t, b.StaticResources.Listeners[0].DefaultFilterChain)
+			require.NotNil(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters)
+			require.Len(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters, 1)
+			require.NotNil(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters[0])
+			var o httpConnectionManagerAPI.HttpConnectionManager
+			tgrpc.GRPCAnyCastAs(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters[0].GetTypedConfig(), &o)
+			rc := o.GetRouteConfig()
+			require.NotNil(t, rc)
+			require.NotNil(t, rc.VirtualHosts)
+			require.Len(t, rc.VirtualHosts, 1)
+			require.NotNil(t, rc.VirtualHosts[0])
+			require.Len(t, rc.VirtualHosts[0].Routes, 1)
+			require.NotNil(t, rc.VirtualHosts[0].Routes[0])
+			r := rc.VirtualHosts[0].Routes[0].GetRoute()
+			require.NotNil(t, r)
+			require.Len(t, r.UpgradeConfigs, 1)
+			require.NotNil(t, r.UpgradeConfigs[0])
+			require.EqualValues(t, "websocket", r.UpgradeConfigs[0].UpgradeType)
+			require.NotNil(t, r.UpgradeConfigs[0].Enabled)
+			require.True(t, r.UpgradeConfigs[0].Enabled.GetValue())
+		})
+	})
+
+	t.Run("With Multi WebSocket", func(t *testing.T) {
+		renderAndPrintGatewayConfig(t, Config{
+			DefaultDestination: ConfigDestination{
+				Targets: []ConfigDestinationTarget{
+					{
+						Host: "127.0.0.1",
+						Port: 12345,
+					},
+				},
+				UpgradeConfigs: ConfigDestinationsUpgrade{
+					{
+						Type: "websocket",
+					},
+					{
+						Type:    "websocket",
+						Enabled: util.NewType(false),
+					},
+				},
+			},
+		}, func(t *testing.T, b *bootstrapAPI.Bootstrap) {
+			require.NotNil(t, b)
+			require.NotNil(t, b.StaticResources)
+			require.NotNil(t, b.StaticResources.Listeners)
+			require.Len(t, b.StaticResources.Listeners, 1)
+			require.NotNil(t, b.StaticResources.Listeners[0])
+			require.NotNil(t, b.StaticResources.Listeners[0].DefaultFilterChain)
+			require.NotNil(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters)
+			require.Len(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters, 1)
+			require.NotNil(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters[0])
+			var o httpConnectionManagerAPI.HttpConnectionManager
+			tgrpc.GRPCAnyCastAs(t, b.StaticResources.Listeners[0].DefaultFilterChain.Filters[0].GetTypedConfig(), &o)
+			rc := o.GetRouteConfig()
+			require.NotNil(t, rc)
+			require.NotNil(t, rc.VirtualHosts)
+			require.Len(t, rc.VirtualHosts, 1)
+			require.NotNil(t, rc.VirtualHosts[0])
+			require.Len(t, rc.VirtualHosts[0].Routes, 1)
+			require.NotNil(t, rc.VirtualHosts[0].Routes[0])
+			r := rc.VirtualHosts[0].Routes[0].GetRoute()
+			require.NotNil(t, r)
+			require.Len(t, r.UpgradeConfigs, 2)
+			require.NotNil(t, r.UpgradeConfigs[0])
+			require.NotNil(t, r.UpgradeConfigs[1])
+			require.EqualValues(t, "websocket", r.UpgradeConfigs[0].UpgradeType)
+			require.NotNil(t, r.UpgradeConfigs[0].Enabled)
+			require.True(t, r.UpgradeConfigs[0].Enabled.GetValue())
+			require.EqualValues(t, "websocket", r.UpgradeConfigs[1].UpgradeType)
+			require.NotNil(t, r.UpgradeConfigs[1].Enabled)
+			require.False(t, r.UpgradeConfigs[1].Enabled.GetValue())
+		})
+	})
 
 	t.Run("Default", func(t *testing.T) {
 		renderAndPrintGatewayConfig(t, Config{

pkg/util/grpc.go

@@ -25,11 +25,13 @@ import (
 	"crypto/tls"
 	"io"
 
+	any1 "github.com/golang/protobuf/ptypes/any"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/metadata"
+	proto "google.golang.org/protobuf/proto"
 
 	pbPongV1 "github.com/arangodb/kube-arangodb/integrations/pong/v1/definition"
 	pbSharedV1 "github.com/arangodb/kube-arangodb/integrations/shared/v1/definition"
@@ -133,3 +135,11 @@ func TokenAuthInterceptors(token string) []grpc.DialOption {
 func attachTokenAuthToInterceptors(ctx context.Context, token string) context.Context {
 	return metadata.AppendToOutgoingContext(ctx, AuthorizationGRPCHeader, token)
 }
+
+func GRPCAnyCastAs[T proto.Message](in *any1.Any, v T) error {
+	if err := in.UnmarshalTo(v); err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/util/tests/tgrpc/grpc.go

@@ -25,10 +25,12 @@ import (
 	"fmt"
 	"testing"
 
+	any1 "github.com/golang/protobuf/ptypes/any"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	proto "google.golang.org/protobuf/proto"
 
 	"github.com/arangodb/kube-arangodb/pkg/util"
 	"github.com/arangodb/kube-arangodb/pkg/util/svc"
@@ -72,3 +74,7 @@ func AsGRPCError(t *testing.T, err error) ErrorStatusValidator {
 	require.NotNil(t, st)
 	return errorStatusValidator{st: st}
 }
+
+func GRPCAnyCastAs[T proto.Message](t *testing.T, in *any1.Any, v T) {
+	require.NoError(t, util.GRPCAnyCastAs[T](in, v))
+}