bump reva

Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de>

changelog/unreleased/bump-reva.md

@@ -0,0 +1,3 @@
+Enhancement: Bump reva
+
+https://github.com/owncloud/ocis/pull/9715

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/coreos/go-oidc/v3 v3.10.0
 	github.com/cs3org/go-cs3apis v0.0.0-20240724121416-062c4e3046cb
-	github.com/cs3org/reva/v2 v2.22.0
+	github.com/cs3org/reva/v2 v2.22.1-0.20240730105121-548644c31544
 	github.com/dhowden/tag v0.0.0-20230630033851-978a0926ee25
 	github.com/dutchcoders/go-clamd v0.0.0-20170520113014-b970184f4d9e
 	github.com/egirna/icap-client v0.1.1
@@ -307,6 +307,7 @@ require (
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/segmentio/kafka-go v0.4.47 // indirect
 	github.com/segmentio/ksuid v1.0.4 // indirect
+	github.com/sercand/kuberesolver/v5 v5.1.1 // indirect
 	github.com/sergi/go-diff v1.3.1 // indirect
 	github.com/sethvargo/go-password v0.2.0 // indirect
 	github.com/shurcooL/httpfs v0.0.0-20190707220628-8d4bc4ba7749 // indirect

go.sum

@@ -1025,8 +1025,8 @@ github.com/crewjam/saml v0.4.14 h1:g9FBNx62osKusnFzs3QTN5L9CVA/Egfgm+stJShzw/c=
 github.com/crewjam/saml v0.4.14/go.mod h1:UVSZCf18jJkk6GpWNVqcyQJMD5HsRugBPf4I1nl2mME=
 github.com/cs3org/go-cs3apis v0.0.0-20240724121416-062c4e3046cb h1:KmYZDReplv/yfwc1LNYpDcVhVujC3Pasv6WjXx1haSU=
 github.com/cs3org/go-cs3apis v0.0.0-20240724121416-062c4e3046cb/go.mod h1:yyP8PRo0EZou3nSH7H4qjlzQwaydPeIRNgX50npQHpE=
-github.com/cs3org/reva/v2 v2.22.0 h1:dMZGJDdrTCYYyMlVSzchbEUyFrHgvFI2Md/aLLiez54=
-github.com/cs3org/reva/v2 v2.22.0/go.mod h1:y9ujkcxepugOsv50baQSCtudfY5VY5S5eigNTvGfSZI=
+github.com/cs3org/reva/v2 v2.22.1-0.20240730105121-548644c31544 h1:cBqx8oou5aXM9SqiG96bYGBD4akYwecPoopsFva51yI=
+github.com/cs3org/reva/v2 v2.22.1-0.20240730105121-548644c31544/go.mod h1:R6OO/ZPMr8MivSiESfk7pUfsdXdr709L8kJErLgqvDI=
 github.com/cyberdelia/templates v0.0.0-20141128023046-ca7fffd4298c/go.mod h1:GyV+0YP4qX0UQ7r2MoYZ+AvYDp12OF5yg4q8rGnyNh4=
 github.com/cyphar/filepath-securejoin v0.2.4 h1:Ugdm7cg7i6ZK6x3xDF1oEu1nfkyfH53EtKeQYTC3kyg=
 github.com/cyphar/filepath-securejoin v0.2.4/go.mod h1:aPGpWjXOXUn2NCNjFvBE6aRxGGx79pTxQpKOJNYHHl4=
@@ -1948,6 +1948,8 @@ github.com/segmentio/kafka-go v0.4.47 h1:IqziR4pA3vrZq7YdRxaT3w1/5fvIH5qpCwstUan
 github.com/segmentio/kafka-go v0.4.47/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg=
 github.com/segmentio/ksuid v1.0.4 h1:sBo2BdShXjmcugAMwjugoGUdUV0pcxY5mW4xKRn3v4c=
 github.com/segmentio/ksuid v1.0.4/go.mod h1:/XUiZBD3kVx5SmUOl55voK5yeAbBNNIed+2O73XgrPE=
+github.com/sercand/kuberesolver/v5 v5.1.1 h1:CYH+d67G0sGBj7q5wLK61yzqJJ8gLLC8aeprPTHb6yY=
+github.com/sercand/kuberesolver/v5 v5.1.1/go.mod h1:Fs1KbKhVRnB2aDWN12NjKCB+RgYMWZJ294T3BtmVCpQ=
 github.com/sergi/go-diff v1.3.1 h1:xkr+Oxo4BOQKmkn/B9eMK0g5Kg/983T9DqqPHwYqD+8=
 github.com/sergi/go-diff v1.3.1/go.mod h1:aMJSSKb2lpPvRNec0+w3fl7LP9IOFzdc9Pa4NFbPK1I=
 github.com/sethgrid/pester v1.2.0/go.mod h1:hEUINb4RqvDxtoCaU0BNT/HV4ig5kfgOasrf1xcvr0A=

vendor/github.com/cs3org/reva/v2/internal/grpc/services/gateway/appprovider.go

@@ -325,12 +325,12 @@ func getGRPCConfig(opaque *typespb.Opaque) (bool, bool) {
 
 func getConn(host string, ins, skipverify bool) (*grpc.ClientConn, error) {
 	if ins {
-		return grpc.Dial(host, grpc.WithTransportCredentials(insecure.NewCredentials()))
+		return grpc.NewClient(host, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	}
 
 	// TODO(labkode): if in the future we want client-side certificate validation,
 	// we need to load the client cert here
 	tlsconf := &tls.Config{InsecureSkipVerify: skipverify}
 	creds := credentials.NewTLS(tlsconf)
-	return grpc.Dial(host, grpc.WithTransportCredentials(creds))
+	return grpc.NewClient(host, grpc.WithTransportCredentials(creds))
 }

vendor/github.com/cs3org/reva/v2/internal/http/services/owncloud/ocdav/copy.go

@@ -292,6 +292,12 @@ func (s *svc) executePathCopy(ctx context.Context, selector pool.Selectable[gate
 			return err
 		}
 		defer httpDownloadRes.Body.Close()
+		if httpDownloadRes.StatusCode == http.StatusForbidden {
+			w.WriteHeader(http.StatusForbidden)
+			b, err := errors.Marshal(http.StatusForbidden, http.StatusText(http.StatusForbidden), "", strconv.Itoa(http.StatusForbidden))
+			errors.HandleWebdavError(log, w, b, err)
+			return nil
+		}
 		if httpDownloadRes.StatusCode != http.StatusOK {
 			return fmt.Errorf("status code %d", httpDownloadRes.StatusCode)
 		}

vendor/github.com/cs3org/reva/v2/pkg/eosclient/eosgrpc/eosgrpc.go

@@ -133,7 +133,7 @@ func newgrpc(ctx context.Context, opt *Options) (erpc.EosClient, error) {
 	log := appctx.GetLogger(ctx)
 	log.Info().Str("Setting up GRPC towards ", "'"+opt.GrpcURI+"'").Msg("")
 
-	conn, err := grpc.Dial(opt.GrpcURI, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	conn, err := grpc.NewClient(opt.GrpcURI, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
 		log.Warn().Str("Error connecting to ", "'"+opt.GrpcURI+"' ").Str("err", err.Error()).Msg("")
 	}

vendor/github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool/connection.go

@@ -38,7 +38,7 @@ var (
 // NewConn creates a new connection to a grpc server
 // with open census tracing support.
 // TODO(labkode): make grpc tls configurable.
-func NewConn(address string, opts ...Option) (*grpc.ClientConn, error) {
+func NewConn(target string, opts ...Option) (*grpc.ClientConn, error) {
 
 	options := ClientOptions{}
 	if err := options.init(); err != nil {
@@ -84,12 +84,34 @@ func NewConn(address string, opts ...Option) (*grpc.ClientConn, error) {
 		maxRcvMsgSize = s
 	}
 
-	conn, err := grpc.Dial(
-		address,
+	conn, err := grpc.NewClient(
+		target,
 		grpc.WithTransportCredentials(cred),
 		grpc.WithDefaultCallOptions(
 			grpc.MaxCallRecvMsgSize(maxRcvMsgSize),
 		),
+		grpc.WithDefaultServiceConfig(`{
+			"loadBalancingPolicy":"round_robin"
+		}`),
+		/* we may want to retry more often than the default transparent retry, see https://grpc.io/docs/guides/retry/#retry-configuration
+		grpc.WithDefaultServiceConfig(`{
+			"loadBalancingPolicy":"round_robin"
+			"methodConfig": [
+				{
+					"name": [
+						{ "service": "grpc.examples.echo.Echo" }
+					],
+					"retryPolicy": {
+						"maxAttempts": 3,
+						"initialBackoff": "0.1s",
+						"maxBackoff": "1s",
+						"backoffMultiplier": 2,
+						"retryableStatusCodes": ["UNAVAILABLE", "CANCELLED", "RESOURCE_EXHAUSTED", "DEADLINE_EXCEEDED"]
+					}
+				}
+			]
+		}`),
+		*/
 		grpc.WithStatsHandler(otelgrpc.NewClientHandler(
 			otelgrpc.WithTracerProvider(
 				options.tracerProvider,

vendor/github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool/selector.go

@@ -20,6 +20,7 @@ package pool
 
 import (
 	"fmt"
+	"strings"
 	"sync"
 
 	appProvider "github.com/cs3org/go-cs3apis/cs3/app/provider/v1beta1"
@@ -43,9 +44,16 @@ import (
 	tx "github.com/cs3org/go-cs3apis/cs3/tx/v1beta1"
 	"github.com/cs3org/reva/v2/pkg/registry"
 	"github.com/pkg/errors"
+	"github.com/sercand/kuberesolver/v5"
 	"google.golang.org/grpc"
 )
 
+func init() {
+	// grpc go resolver.Register must only be called during initialization time (i.e. in
+	// an init() function), and is not thread-safe.
+	kuberesolver.RegisterInCluster()
+}
+
 type Selectable[T any] interface {
 	Next(opts ...Option) (T, error)
 }
@@ -93,8 +101,19 @@ func (s *Selector[T]) Next(opts ...Option) (T, error) {
 		opt(&options)
 	}
 
-	address := s.id
-	if options.registry != nil {
+	target := s.id
+	// if the target is given as a recognized gRPC URI, skip registry lookup
+	// see https://github.com/grpc/grpc/blob/master/doc/naming.md#name-syntax
+	prefix := strings.SplitN(s.id, ":", 2)[0]
+	switch {
+	case prefix == "dns":
+		fallthrough
+	case prefix == "unix":
+		fallthrough
+	case prefix == "kubernetes":
+		// use target as is and skip registry lookup
+	case options.registry != nil:
+		// use service registry to look up address
 		services, err := options.registry.GetService(s.id)
 		if err != nil {
 			return *new(T), fmt.Errorf("%s: %w", s.id, err)
@@ -104,22 +123,23 @@ func (s *Selector[T]) Next(opts ...Option) (T, error) {
 		if err != nil {
 			return *new(T), fmt.Errorf("%s: %w", s.id, err)
 		}
-
-		address = nodeAddress
+		target = nodeAddress
+	default:
+		// if no registry is available, use the target as is
 	}
 
-	existingClient, ok := s.clientMap.Load(address)
+	existingClient, ok := s.clientMap.Load(target)
 	if ok {
 		return existingClient.(T), nil
 	}
 
-	conn, err := NewConn(address, allOpts...)
+	conn, err := NewConn(target, allOpts...)
 	if err != nil {
-		return *new(T), errors.Wrap(err, fmt.Sprintf("could not create connection for %s to %s", s.id, address))
+		return *new(T), errors.Wrap(err, fmt.Sprintf("could not create connection for %s to %s", s.id, target))
 	}
 
 	newClient := s.clientFactory(conn)
-	s.clientMap.Store(address, newClient)
+	s.clientMap.Store(target, newClient)
 
 	return newClient, nil
 }

vendor/github.com/cs3org/reva/v2/pkg/trace/trace.go

@@ -22,7 +22,6 @@ import (
 	"context"
 	"fmt"
 	"sync"
-	"time"
 
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
@@ -106,13 +105,11 @@ func DefaultProvider() trace.TracerProvider {
 
 // getOtelTracerProvider returns a new TracerProvider, configure for the specified service
 func getOtlpTracerProvider(options Options) trace.TracerProvider {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-	defer cancel()
 	transportCredentials := options.TransportCredentials
 	if options.Insecure {
 		transportCredentials = insecure.NewCredentials()
 	}
-	conn, err := grpc.DialContext(ctx, options.Endpoint,
+	conn, err := grpc.NewClient(options.Endpoint,
 		grpc.WithTransportCredentials(transportCredentials),
 	)
 	if err != nil {

vendor/github.com/sercand/kuberesolver/v5/.gitignore

@@ -0,0 +1,2 @@
+.idea
+kuberesolver.iml

vendor/github.com/sercand/kuberesolver/v5/LICENSE

@@ -0,0 +1,202 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright 2017 Sercan Degirmenci
+
+   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.

vendor/github.com/sercand/kuberesolver/v5/README.md

@@ -0,0 +1,68 @@
+# kuberesolver
+
+A Grpc name resolver by using kubernetes API.
+It comes with a small ~250 LOC kubernetes client to find service endpoints. Therefore it won't bloat your binaries.
+
+
+### USAGE
+
+```go
+
+// Import the module
+import "github.com/sercand/kuberesolver/v5"
+	
+// Register kuberesolver to grpc before calling grpc.Dial
+kuberesolver.RegisterInCluster()
+
+// it is same as
+resolver.Register(kuberesolver.NewBuilder(nil /*custom kubernetes client*/ , "kubernetes"))
+
+// if schema is 'kubernetes' then grpc will use kuberesolver to resolve addresses
+cc, err := grpc.Dial("kubernetes:///service.namespace:portname", opts...)
+```
+
+An url can be one of the following, [grpc naming docs](https://github.com/grpc/grpc/blob/master/doc/naming.md)
+
+```
+kubernetes:///service-name:8080
+kubernetes:///service-name:portname
+kubernetes:///service-name.namespace:8080
+kubernetes:///service-name.namespace.svc.cluster_name
+kubernetes:///service-name.namespace.svc.cluster_name:8080
+
+kubernetes://namespace/service-name:8080
+kubernetes://service-name:8080/
+kubernetes://service-name.namespace:8080/
+kubernetes://service-name.namespace.svc.cluster_name
+kubernetes://service-name.namespace.svc.cluster_name:8080
+```
+_* Please note that the cluster_name is not used in resolving the endpoints of a Service. It is only there to support fully qualified service names, e.g._ `test.default.svc.cluster.local`.
+
+### Using alternative Schema
+
+Use `RegisterInClusterWithSchema(schema)` instead of `RegisterInCluster` on start.
+
+### Client Side Load Balancing
+
+You need to pass grpc.WithBalancerName option to grpc on dial: 
+
+```go
+grpc.DialContext(ctx,  "kubernetes:///service:grpc", grpc.WithBalancerName("round_robin"), grpc.WithInsecure())
+```
+This will create subconnections for each available service endpoints.
+
+### How is this different from dialing to `service.namespace:8080`
+
+Connecting to a service by dialing to `service.namespace:8080` uses DNS and it returns service stable IP. Therefore, gRPC doesn't know the endpoint IP addresses and it fails to reconnect to target services in case of failure.  
+
+Kuberesolver uses kubernetes API to get and watch service endpoint IP addresses. 
+Since it provides and updates all available service endpoints, together with a client-side balancer you can achive zero downtime deployments.
+
+### RBAC
+
+You need give `GET` and `WATCH` access to the `endpoints` if you are using RBAC in your cluster.
+
+
+### Using With TLS
+
+You need to a certificate with name `service-name.namespace` in order to connect with TLS to your services.

vendor/github.com/sercand/kuberesolver/v5/builder.go

@@ -0,0 +1,289 @@
+package kuberesolver
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"google.golang.org/grpc/grpclog"
+	"google.golang.org/grpc/resolver"
+)
+
+const (
+	kubernetesSchema = "kubernetes"
+	defaultFreq      = time.Minute * 30
+)
+
+var (
+	endpointsForTarget = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "kuberesolver_endpoints_total",
+			Help: "The number of endpoints for a given target",
+		},
+		[]string{"target"},
+	)
+	addressesForTarget = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "kuberesolver_addresses_total",
+			Help: "The number of addresses for a given target",
+		},
+		[]string{"target"},
+	)
+)
+
+type targetInfo struct {
+	serviceName       string
+	serviceNamespace  string
+	port              string
+	resolveByPortName bool
+	useFirstPort      bool
+}
+
+func (ti targetInfo) String() string {
+	return fmt.Sprintf("kubernetes://%s/%s:%s", ti.serviceNamespace, ti.serviceName, ti.port)
+}
+
+// RegisterInCluster registers the kuberesolver builder to grpc with kubernetes schema
+func RegisterInCluster() {
+	RegisterInClusterWithSchema(kubernetesSchema)
+}
+
+// RegisterInClusterWithSchema registers the kuberesolver builder to the grpc with custom schema
+func RegisterInClusterWithSchema(schema string) {
+	resolver.Register(NewBuilder(nil, schema))
+}
+
+// NewBuilder creates a kubeBuilder which is used by grpc resolver.
+func NewBuilder(client K8sClient, schema string) resolver.Builder {
+	return &kubeBuilder{
+		k8sClient: client,
+		schema:    schema,
+	}
+}
+
+type kubeBuilder struct {
+	k8sClient K8sClient
+	schema    string
+}
+
+func splitServicePortNamespace(hpn string) (service, port, namespace string) {
+	service = hpn
+
+	colon := strings.LastIndexByte(service, ':')
+	if colon != -1 {
+		service, port = service[:colon], service[colon+1:]
+	}
+
+	// we want to split into the service name, namespace, and whatever else is left
+	// this will support fully qualified service names, e.g. {service-name}.<namespace>.svc.<cluster-domain-name>.
+	// Note that since we lookup the endpoints by service name and namespace, we don't care about the
+	// cluster-domain-name, only that we can parse out the service name and namespace properly.
+	parts := strings.SplitN(service, ".", 3)
+	if len(parts) >= 2 {
+		service, namespace = parts[0], parts[1]
+	}
+
+	return
+}
+
+func parseResolverTarget(target resolver.Target) (targetInfo, error) {
+	var service, port, namespace string
+	if target.URL.Host == "" {
+		// kubernetes:///service.namespace:port
+		service, port, namespace = splitServicePortNamespace(target.Endpoint())
+	} else if target.URL.Port() == "" && target.Endpoint() != "" {
+		// kubernetes://namespace/service:port
+		service, port, _ = splitServicePortNamespace(target.Endpoint())
+		namespace = target.URL.Hostname()
+	} else {
+		// kubernetes://service.namespace:port
+		service, port, namespace = splitServicePortNamespace(target.URL.Host)
+	}
+
+	if service == "" {
+		return targetInfo{}, fmt.Errorf("target %s must specify a service", &target.URL)
+	}
+
+	resolveByPortName := false
+	useFirstPort := false
+	if port == "" {
+		useFirstPort = true
+	} else if _, err := strconv.Atoi(port); err != nil {
+		resolveByPortName = true
+	}
+
+	return targetInfo{
+		serviceName:       service,
+		serviceNamespace:  namespace,
+		port:              port,
+		resolveByPortName: resolveByPortName,
+		useFirstPort:      useFirstPort,
+	}, nil
+}
+
+// Build creates a new resolver for the given target.
+//
+// gRPC dial calls Build synchronously, and fails if the returned error is
+// not nil.
+func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
+	if b.k8sClient == nil {
+		if cl, err := NewInClusterK8sClient(); err == nil {
+			b.k8sClient = cl
+		} else {
+			return nil, err
+		}
+	}
+	ti, err := parseResolverTarget(target)
+	if err != nil {
+		return nil, err
+	}
+	if ti.serviceNamespace == "" {
+		ti.serviceNamespace = getCurrentNamespaceOrDefault()
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	r := &kResolver{
+		target:    ti,
+		ctx:       ctx,
+		cancel:    cancel,
+		cc:        cc,
+		rn:        make(chan struct{}, 1),
+		k8sClient: b.k8sClient,
+		t:         time.NewTimer(defaultFreq),
+		freq:      defaultFreq,
+
+		endpoints: endpointsForTarget.WithLabelValues(ti.String()),
+		addresses: addressesForTarget.WithLabelValues(ti.String()),
+	}
+	go until(func() {
+		r.wg.Add(1)
+		err := r.watch()
+		if err != nil && err != io.EOF {
+			grpclog.Errorf("kuberesolver: watching ended with error='%v', will reconnect again", err)
+		}
+	}, time.Second, time.Second*30, ctx.Done())
+	return r, nil
+}
+
+// Scheme returns the scheme supported by this resolver.
+// Scheme is defined at https://github.com/grpc/grpc/blob/master/doc/naming.md.
+func (b *kubeBuilder) Scheme() string {
+	return b.schema
+}
+
+type kResolver struct {
+	target targetInfo
+	ctx    context.Context
+	cancel context.CancelFunc
+	cc     resolver.ClientConn
+	// rn channel is used by ResolveNow() to force an immediate resolution of the target.
+	rn        chan struct{}
+	k8sClient K8sClient
+	// wg is used to enforce Close() to return after the watcher() goroutine has finished.
+	wg   sync.WaitGroup
+	t    *time.Timer
+	freq time.Duration
+
+	endpoints prometheus.Gauge
+	addresses prometheus.Gauge
+}
+
+// ResolveNow will be called by gRPC to try to resolve the target name again.
+// It's just a hint, resolver can ignore this if it's not necessary.
+func (k *kResolver) ResolveNow(resolver.ResolveNowOptions) {
+	select {
+	case k.rn <- struct{}{}:
+	default:
+	}
+}
+
+// Close closes the resolver.
+func (k *kResolver) Close() {
+	k.cancel()
+	k.wg.Wait()
+}
+
+func (k *kResolver) makeAddresses(e Endpoints) ([]resolver.Address, string) {
+	var newAddrs []resolver.Address
+	for _, subset := range e.Subsets {
+		port := ""
+		if k.target.useFirstPort {
+			port = strconv.Itoa(subset.Ports[0].Port)
+		} else if k.target.resolveByPortName {
+			for _, p := range subset.Ports {
+				if p.Name == k.target.port {
+					port = strconv.Itoa(p.Port)
+					break
+				}
+			}
+		} else {
+			port = k.target.port
+		}
+
+		if len(port) == 0 {
+			port = strconv.Itoa(subset.Ports[0].Port)
+		}
+
+		for _, address := range subset.Addresses {
+			newAddrs = append(newAddrs, resolver.Address{
+				Addr:       net.JoinHostPort(address.IP, port),
+				ServerName: fmt.Sprintf("%s.%s", k.target.serviceName, k.target.serviceNamespace),
+				Metadata:   nil,
+			})
+		}
+	}
+	return newAddrs, ""
+}
+
+func (k *kResolver) handle(e Endpoints) {
+	result, _ := k.makeAddresses(e)
+	//	k.cc.NewServiceConfig(sc)
+	if len(result) > 0 {
+		k.cc.NewAddress(result)
+	}
+
+	k.endpoints.Set(float64(len(e.Subsets)))
+	k.addresses.Set(float64(len(result)))
+}
+
+func (k *kResolver) resolve() {
+	e, err := getEndpoints(k.k8sClient, k.target.serviceNamespace, k.target.serviceName)
+	if err == nil {
+		k.handle(e)
+	} else {
+		grpclog.Errorf("kuberesolver: lookup endpoints failed: %v", err)
+	}
+	// Next lookup should happen after an interval defined by k.freq.
+	k.t.Reset(k.freq)
+}
+
+func (k *kResolver) watch() error {
+	defer k.wg.Done()
+	// watch endpoints lists existing endpoints at start
+	sw, err := watchEndpoints(k.ctx, k.k8sClient, k.target.serviceNamespace, k.target.serviceName)
+	if err != nil {
+		return err
+	}
+	for {
+		select {
+		case <-k.ctx.Done():
+			return nil
+		case <-k.t.C:
+			k.resolve()
+		case <-k.rn:
+			//k.resolve()
+		case up, hasMore := <-sw.ResultChan():
+			if hasMore {
+				k.handle(up.Object)
+			} else {
+				return nil
+			}
+		}
+	}
+}

vendor/github.com/sercand/kuberesolver/v5/kubernetes.go

@@ -0,0 +1,206 @@
+package kuberesolver
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"fmt"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/fsnotify/fsnotify"
+)
+
+const (
+	serviceAccountToken     = "/var/run/secrets/kubernetes.io/serviceaccount/token"
+	serviceAccountCACert    = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+	kubernetesNamespaceFile = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
+	defaultNamespace        = "default"
+)
+
+// K8sClient is minimal kubernetes client interface
+type K8sClient interface {
+	Do(req *http.Request) (*http.Response, error)
+	GetRequest(url string) (*http.Request, error)
+	Host() string
+}
+
+type k8sClient struct {
+	host       string
+	token      string
+	tokenLck   sync.RWMutex
+	httpClient *http.Client
+}
+
+func (kc *k8sClient) GetRequest(url string) (*http.Request, error) {
+	if !strings.HasPrefix(url, kc.host) {
+		url = fmt.Sprintf("%s/%s", kc.host, url)
+	}
+	req, err := http.NewRequest("GET", url, nil)
+	if err != nil {
+		return nil, err
+	}
+	kc.tokenLck.RLock()
+	defer kc.tokenLck.RUnlock()
+	if len(kc.token) > 0 {
+		req.Header.Set("Authorization", "Bearer "+kc.token)
+	}
+	return req, nil
+}
+
+func (kc *k8sClient) Do(req *http.Request) (*http.Response, error) {
+	return kc.httpClient.Do(req)
+}
+
+func (kc *k8sClient) Host() string {
+	return kc.host
+}
+
+func (kc *k8sClient) setToken(token string) {
+	kc.tokenLck.Lock()
+	defer kc.tokenLck.Unlock()
+	kc.token = token
+}
+
+// NewInClusterK8sClient creates K8sClient if it is inside Kubernetes
+func NewInClusterK8sClient() (K8sClient, error) {
+	host, port := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT")
+	if len(host) == 0 || len(port) == 0 {
+		return nil, fmt.Errorf("unable to load in-cluster configuration, KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT must be defined")
+	}
+	token, err := os.ReadFile(serviceAccountToken)
+	if err != nil {
+		return nil, err
+	}
+	ca, err := os.ReadFile(serviceAccountCACert)
+	if err != nil {
+		return nil, err
+	}
+	certPool := x509.NewCertPool()
+	certPool.AppendCertsFromPEM(ca)
+	transport := &http.Transport{TLSClientConfig: &tls.Config{
+		MinVersion: tls.VersionTLS10,
+		RootCAs:    certPool,
+	}}
+	httpClient := &http.Client{Transport: transport, Timeout: time.Nanosecond * 0}
+
+	client := &k8sClient{
+		host:       "https://" + net.JoinHostPort(host, port),
+		token:      string(token),
+		httpClient: httpClient,
+	}
+
+	// Create a new file watcher to listen for new Service Account tokens
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		for {
+			select {
+			case event, ok := <-watcher.Events:
+				if !ok {
+					return
+				}
+				// k8s configmaps uses symlinks, we need this workaround.
+				// original configmap file is removed
+				if event.Op.Has(fsnotify.Remove) || event.Op.Has(fsnotify.Chmod) {
+					// remove watcher since the file is removed
+					watcher.Remove(event.Name)
+					// add a new watcher pointing to the new symlink/file
+					watcher.Add(serviceAccountToken)
+					token, err := os.ReadFile(serviceAccountToken)
+					if err == nil {
+						client.setToken(string(token))
+					}
+				}
+				if event.Has(fsnotify.Write) {
+					token, err := os.ReadFile(serviceAccountToken)
+					if err == nil {
+						client.setToken(string(token))
+					}
+				}
+			case _, ok := <-watcher.Errors:
+				if !ok {
+					return
+				}
+			}
+		}
+	}()
+
+	err = watcher.Add(serviceAccountToken)
+	if err != nil {
+		return nil, err
+	}
+
+	return client, nil
+}
+
+// NewInsecureK8sClient creates an insecure k8s client which is suitable
+// to connect kubernetes api behind proxy
+func NewInsecureK8sClient(apiURL string) K8sClient {
+	return &k8sClient{
+		host:       apiURL,
+		httpClient: http.DefaultClient,
+	}
+}
+
+func getEndpoints(client K8sClient, namespace, targetName string) (Endpoints, error) {
+	u, err := url.Parse(fmt.Sprintf("%s/api/v1/namespaces/%s/endpoints/%s",
+		client.Host(), namespace, targetName))
+	if err != nil {
+		return Endpoints{}, err
+	}
+	req, err := client.GetRequest(u.String())
+	if err != nil {
+		return Endpoints{}, err
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		return Endpoints{}, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return Endpoints{}, fmt.Errorf("invalid response code %d for service %s in namespace %s", resp.StatusCode, targetName, namespace)
+	}
+	result := Endpoints{}
+	err = json.NewDecoder(resp.Body).Decode(&result)
+	return result, err
+}
+
+func watchEndpoints(ctx context.Context, client K8sClient, namespace, targetName string) (watchInterface, error) {
+	u, err := url.Parse(fmt.Sprintf("%s/api/v1/watch/namespaces/%s/endpoints/%s",
+		client.Host(), namespace, targetName))
+	if err != nil {
+		return nil, err
+	}
+	req, err := client.GetRequest(u.String())
+	if err != nil {
+		return nil, err
+	}
+	req = req.WithContext(ctx)
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		defer resp.Body.Close()
+		return nil, fmt.Errorf("invalid response code %d for service %s in namespace %s", resp.StatusCode, targetName, namespace)
+	}
+	return newStreamWatcher(resp.Body), nil
+}
+
+func getCurrentNamespaceOrDefault() string {
+	ns, err := os.ReadFile(kubernetesNamespaceFile)
+	if err != nil {
+		return defaultNamespace
+	}
+	return string(ns)
+}

vendor/github.com/sercand/kuberesolver/v5/models.go

@@ -0,0 +1,50 @@
+package kuberesolver
+
+type EventType string
+
+const (
+	Added    EventType = "ADDED"
+	Modified EventType = "MODIFIED"
+	Deleted  EventType = "DELETED"
+	Error    EventType = "ERROR"
+)
+
+// Event represents a single event to a watched resource.
+type Event struct {
+	Type   EventType `json:"type"`
+	Object Endpoints `json:"object"`
+}
+
+type Endpoints struct {
+	Kind       string   `json:"kind"`
+	ApiVersion string   `json:"apiVersion"`
+	Metadata   Metadata `json:"metadata"`
+	Subsets    []Subset `json:"subsets"`
+}
+
+type Metadata struct {
+	Name            string            `json:"name"`
+	Namespace       string            `json:"namespace"`
+	ResourceVersion string            `json:"resourceVersion"`
+	Labels          map[string]string `json:"labels"`
+}
+
+type Subset struct {
+	Addresses []Address `json:"addresses"`
+	Ports     []Port    `json:"ports"`
+}
+
+type Address struct {
+	IP        string           `json:"ip"`
+	TargetRef *ObjectReference `json:"targetRef,omitempty"`
+}
+
+type ObjectReference struct {
+	Kind      string `json:"kind"`
+	Name      string `json:"name"`
+	Namespace string `json:"namespace"`
+}
+type Port struct {
+	Name string `json:"name"`
+	Port int    `json:"port"`
+}

vendor/github.com/sercand/kuberesolver/v5/stream.go

@@ -0,0 +1,108 @@
+package kuberesolver
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"sync"
+
+	"google.golang.org/grpc/grpclog"
+)
+
+// Interface can be implemented by anything that knows how to watch and report changes.
+type watchInterface interface {
+	// Stops watching. Will close the channel returned by ResultChan(). Releases
+	// any resources used by the watch.
+	Stop()
+
+	// Returns a chan which will receive all the events. If an error occurs
+	// or Stop() is called, this channel will be closed, in which case the
+	// watch should be completely cleaned up.
+	ResultChan() <-chan Event
+}
+
+// StreamWatcher turns any stream for which you can write a Decoder interface
+// into a watch.Interface.
+type streamWatcher struct {
+	result  chan Event
+	r       io.ReadCloser
+	decoder *json.Decoder
+	sync.Mutex
+	stopped bool
+}
+
+// NewStreamWatcher creates a StreamWatcher from the given io.ReadClosers.
+func newStreamWatcher(r io.ReadCloser) watchInterface {
+	sw := &streamWatcher{
+		r:       r,
+		decoder: json.NewDecoder(r),
+		result:  make(chan Event),
+	}
+	go sw.receive()
+	return sw
+}
+
+// ResultChan implements Interface.
+func (sw *streamWatcher) ResultChan() <-chan Event {
+	return sw.result
+}
+
+// Stop implements Interface.
+func (sw *streamWatcher) Stop() {
+	sw.Lock()
+	defer sw.Unlock()
+	if !sw.stopped {
+		sw.stopped = true
+		sw.r.Close()
+	}
+}
+
+// stopping returns true if Stop() was called previously.
+func (sw *streamWatcher) stopping() bool {
+	sw.Lock()
+	defer sw.Unlock()
+	return sw.stopped
+}
+
+// receive reads result from the decoder in a loop and sends down the result channel.
+func (sw *streamWatcher) receive() {
+	defer close(sw.result)
+	defer sw.Stop()
+	for {
+		obj, err := sw.Decode()
+		if err != nil {
+			// Ignore expected error.
+			if sw.stopping() {
+				return
+			}
+			switch err {
+			case io.EOF:
+				// watch closed normally
+			case context.Canceled:
+				// canceled normally
+			case io.ErrUnexpectedEOF:
+				grpclog.Infof("kuberesolver: Unexpected EOF during watch stream event decoding: %v", err)
+			default:
+				grpclog.Infof("kuberesolver: Unable to decode an event from the watch stream: %v", err)
+			}
+			return
+		}
+		sw.result <- obj
+	}
+}
+
+// Decode blocks until it can return the next object in the writer. Returns an error
+// if the writer is closed or an object can't be decoded.
+func (sw *streamWatcher) Decode() (Event, error) {
+	var got Event
+	if err := sw.decoder.Decode(&got); err != nil {
+		return Event{}, err
+	}
+	switch got.Type {
+	case Added, Modified, Deleted, Error:
+		return got, nil
+	default:
+		return Event{}, fmt.Errorf("got invalid watch event type: %v", got.Type)
+	}
+}

vendor/github.com/sercand/kuberesolver/v5/util.go

@@ -0,0 +1,41 @@
+package kuberesolver
+
+import (
+	"runtime/debug"
+	"time"
+
+	"google.golang.org/grpc/grpclog"
+)
+
+func until(f func(), initialPeriod, maxPeriod time.Duration, stopCh <-chan struct{}) {
+	select {
+	case <-stopCh:
+		return
+	default:
+	}
+	period := initialPeriod
+	for {
+		func() {
+			defer handleCrash()
+			f()
+		}()
+		select {
+		case <-stopCh:
+			return
+		case <-time.After(period):
+			if period*2 <= maxPeriod {
+				period *= 2
+			} else {
+				period = initialPeriod
+			}
+		}
+	}
+}
+
+// HandleCrash simply catches a crash and logs an error. Meant to be called via defer.
+func handleCrash() {
+	if r := recover(); r != nil {
+		callers := string(debug.Stack())
+		grpclog.Errorf("kuberesolver: recovered from panic: %#v (%v)\n%v", r, r, callers)
+	}
+}

vendor/modules.txt

@@ -366,7 +366,7 @@ github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1
 github.com/cs3org/go-cs3apis/cs3/storage/registry/v1beta1
 github.com/cs3org/go-cs3apis/cs3/tx/v1beta1
 github.com/cs3org/go-cs3apis/cs3/types/v1beta1
-# github.com/cs3org/reva/v2 v2.22.0
+# github.com/cs3org/reva/v2 v2.22.1-0.20240730105121-548644c31544
 ## explicit; go 1.21
 github.com/cs3org/reva/v2/cmd/revad/internal/grace
 github.com/cs3org/reva/v2/cmd/revad/runtime
@@ -1785,6 +1785,9 @@ github.com/segmentio/kafka-go/sasl
 # github.com/segmentio/ksuid v1.0.4
 ## explicit; go 1.12
 github.com/segmentio/ksuid
+# github.com/sercand/kuberesolver/v5 v5.1.1
+## explicit; go 1.18
+github.com/sercand/kuberesolver/v5
 # github.com/sergi/go-diff v1.3.1
 ## explicit; go 1.12
 github.com/sergi/go-diff/diffmatchpatch