From 943c2357a9b0006fe7c10b584c72243372fab490 Mon Sep 17 00:00:00 2001 From: Gianmaria Del Monte Date: Fri, 28 Aug 2026 16:23:40 +0200 Subject: [PATCH 1/5] Extract the invocation fan-out into pkg/invoke/client Selector resolution and control-channel dialing move out of the admin service so other fleet-internal consumers can invoke services too. The admin service delegates to it; behavior is unchanged. --- internal/grpc/services/admin/fanout_test.go | 8 +- internal/grpc/services/admin/invoke.go | 188 ++----------- internal/grpc/services/admin/jobs.go | 9 +- .../grpc/services/admin/orchestrate_test.go | 43 +-- internal/grpc/services/admin/peer.go | 57 ---- internal/grpc/services/admin/read.go | 3 +- internal/grpc/services/admin/stream.go | 23 +- pkg/invoke/client/client.go | 250 ++++++++++++++++++ 8 files changed, 314 insertions(+), 267 deletions(-) delete mode 100644 internal/grpc/services/admin/peer.go create mode 100644 pkg/invoke/client/client.go diff --git a/internal/grpc/services/admin/fanout_test.go b/internal/grpc/services/admin/fanout_test.go index f9f2ba9bef3..2b90cf9f1aa 100644 --- a/internal/grpc/services/admin/fanout_test.go +++ b/internal/grpc/services/admin/fanout_test.go @@ -21,14 +21,16 @@ package admin import ( "context" "testing" + + "github.com/cs3org/reva/v3/pkg/invoke/client" ) // TestFanOutInvokeReportsPerNodeErrors checks that unresolved/offline nodes are // reported per-node (in order) rather than failing the whole fan-out. func TestFanOutInvokeReportsPerNodeErrors(t *testing.T) { - eps := []endpoint{ - {node: "n1", err: "node advertises no control endpoint"}, - {node: "n2", err: "offline"}, + eps := []client.Endpoint{ + {Node: "n1", Err: "node advertises no control endpoint"}, + {Node: "n2", Err: "offline"}, } res := fanOutInvoke(context.Background(), eps, "op", nil) if len(res) != 2 { diff --git a/internal/grpc/services/admin/invoke.go b/internal/grpc/services/admin/invoke.go index f6263805ea5..0f100e9be12 100644 --- a/internal/grpc/services/admin/invoke.go +++ b/internal/grpc/services/admin/invoke.go @@ -21,33 +21,20 @@ package admin import ( "context" "fmt" - "net" - "sort" "strconv" "strings" - "sync" - "time" "github.com/cs3org/reva/v3/pkg/admin" "github.com/cs3org/reva/v3/pkg/admin/adminpb" "github.com/cs3org/reva/v3/pkg/appctx" "github.com/cs3org/reva/v3/pkg/control/controlpb" "github.com/cs3org/reva/v3/pkg/invoke" + "github.com/cs3org/reva/v3/pkg/invoke/client" "github.com/cs3org/reva/v3/pkg/registry" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) -// endpoint is one resolved invocation target: the node id to report, the -// control address to dial, and the target the control channel routes on. err -// set means the target could not be resolved. -type endpoint struct { - node string - addr string - target string - err string -} - // ListInvocations returns the invocations a service exposes: the full specs // from one live instance's control channel, falling back to the names in // registry metadata if none is reachable. @@ -59,19 +46,19 @@ func (s *svc) ListInvocations(ctx context.Context, req *adminpb.ListInvocationsR if err != nil { return nil, err } - svcName, eps, err := resolveSelector(reg, req.Service) + svcName, eps, err := client.Resolve(reg, req.Service) if err != nil { return nil, status.Errorf(codes.NotFound, "admin: %v", err) } for _, ep := range eps { - if ep.addr == "" { + if ep.Addr == "" { continue } - cli, err := controlClientAt(ep.addr) + cli, err := client.ControlClientAt(ep.Addr) if err != nil { continue } - if resp, err := cli.ListInvocations(ctx, &controlpb.ListInvocationsRequest{Target: ep.target}); err == nil { + if resp, err := cli.ListInvocations(ctx, &controlpb.ListInvocationsRequest{Target: ep.Target}); err == nil { return &adminpb.ListInvocationsResponse{Invocations: specsToAdmin(resp.Invocations)}, nil } } @@ -95,7 +82,7 @@ func (s *svc) Invoke(ctx context.Context, req *adminpb.InvokeRequest) (*adminpb. if err != nil { return nil, err } - _, eps, err := resolveSelector(reg, req.Service) + _, eps, err := client.Resolve(reg, req.Service) if err != nil { return nil, status.Errorf(codes.NotFound, "admin: resolving %q: %v", req.Service, err) } @@ -105,162 +92,17 @@ func (s *svc) Invoke(ctx context.Context, req *adminpb.InvokeRequest) (*adminpb. return &adminpb.InvokeResponse{Results: results}, nil } -// perNodeTimeout bounds a single peer invocation so one slow or offline node -// never stalls a fleet-wide fan-out. -const perNodeTimeout = 10 * time.Second - // fanOutInvoke invokes every endpoint in parallel. An unreachable node is a // per-node error rather than a failure of the whole call. -func fanOutInvoke(ctx context.Context, eps []endpoint, invocation string, args map[string]string) []*adminpb.NodeResult { - results := make([]*adminpb.NodeResult, len(eps)) - var wg sync.WaitGroup - for i, ep := range eps { - wg.Add(1) - go func(i int, ep endpoint) { - defer wg.Done() - results[i] = invokeOne(ctx, ep, invocation, args) - }(i, ep) +func fanOutInvoke(ctx context.Context, eps []client.Endpoint, invocation string, args map[string]string) []*adminpb.NodeResult { + rs := client.FanOut(ctx, eps, invocation, args) + results := make([]*adminpb.NodeResult, len(rs)) + for i, r := range rs { + results[i] = &adminpb.NodeResult{Node: r.Node, ResultJson: r.ResultJSON, Error: r.Error} } - wg.Wait() return results } -// invokeOne runs a single peer invocation with a bounded timeout. -func invokeOne(ctx context.Context, ep endpoint, invocation string, args map[string]string) *adminpb.NodeResult { - if ep.err != "" { - return &adminpb.NodeResult{Node: ep.node, Error: ep.err} - } - cli, err := controlClientAt(ep.addr) - if err != nil { - return &adminpb.NodeResult{Node: ep.node, Error: err.Error()} - } - cctx, cancel := context.WithTimeout(ctx, perNodeTimeout) - defer cancel() - resp, err := cli.Invoke(cctx, &controlpb.InvokeRequest{Target: ep.target, Invocation: invocation, Args: args}) - if err != nil { - return &adminpb.NodeResult{Node: ep.node, Error: err.Error()} - } - return &adminpb.NodeResult{Node: ep.node, ResultJson: resp.ResultJson, Error: resp.Error} -} - -// resolveSelector maps a selector to its control endpoints: a node id -// "host:port/service" targets one instance, a service name every live one, a -// partial id ("host:port" or a bare host) every instance at that address or on -// that machine, and "*" every live instance in the fleet. -func resolveSelector(reg registry.Registry, selector string) (string, []endpoint, error) { - // "*": every live instance in the fleet. - if selector == "*" { - eps := endpointsMatching(reg, func(registry.Node) bool { return true }) - if len(eps) == 0 { - return "", nil, fmt.Errorf("no live instances in the fleet") - } - return selector, eps, nil - } - - // Node id "host:port/service": one exact instance. - if i := strings.LastIndex(selector, "/"); i >= 0 { - svcName := selector[i+1:] - if svcName == "" { - return "", nil, fmt.Errorf("invalid instance id %q", selector) - } - sv, err := reg.GetService(svcName) - if err != nil { - return "", nil, fmt.Errorf("instance %q: service %q not found", selector, svcName) - } - for _, n := range sv.Nodes() { - if n.ID() == selector { - return svcName, []endpoint{controlEndpointFor(n)}, nil - } - } - return "", nil, fmt.Errorf("instance %q not found", selector) - } - - // Plain service name: every live instance. - if sv, err := reg.GetService(selector); err == nil && len(sv.Nodes()) > 0 { - var eps []endpoint - for _, n := range sv.Nodes() { - // A drained node is out of service rotation but still alive and - // control-reachable — keep it so it can be enabled again (and so - // logs/stack/config still work against it). Only offline is skipped. - if nodeState(n) == registry.StateOffline { - continue - } - eps = append(eps, controlEndpointFor(n)) - } - if len(eps) == 0 { - return "", nil, fmt.Errorf("service %q has no live instances", selector) - } - return selector, eps, nil - } - - // Partial id: "host:port" targets every instance at that address, a bare - // host every instance on that machine. - if eps := endpointsMatchingAddress(reg, selector); len(eps) > 0 { - return selector, eps, nil - } - - return "", nil, fmt.Errorf("%q matches no service, instance, address or host", selector) -} - -// endpointsMatchingAddress resolves a partial node id: "host:port" matches the -// live instances bound to that address, a bare host those on that machine (by -// the id's host part or the node's host metadata). -func endpointsMatchingAddress(reg registry.Registry, selector string) []endpoint { - byAddress := strings.Contains(selector, ":") - return endpointsMatching(reg, func(n registry.Node) bool { - if byAddress { - return strings.HasPrefix(n.ID(), selector+"/") - } - return onHost(n, selector) - }) -} - -// endpointsMatching gathers the live instances accepted by match, sorted by -// node id. -func endpointsMatching(reg registry.Registry, match func(registry.Node) bool) []endpoint { - svcs, err := reg.ListServices() - if err != nil { - return nil - } - var eps []endpoint - for _, sv := range svcs { - for _, n := range sv.Nodes() { - // Drained nodes stay reachable for control (see resolveSelector); - // only offline is skipped. - if nodeState(n) == registry.StateOffline { - continue - } - if !match(n) { - continue - } - eps = append(eps, controlEndpointFor(n)) - } - } - sort.Slice(eps, func(i, j int) bool { return eps[i].node < eps[j].node }) - return eps -} - -// onHost reports whether a node runs on the given host, by the host part of its -// id's address or by its host metadata. -func onHost(n registry.Node, host string) bool { - id := n.ID() - if i := strings.LastIndex(id, "/"); i >= 0 { - if h, _, err := net.SplitHostPort(id[:i]); err == nil && h == host { - return true - } - } - return n.Metadata()["host"] == host -} - -// controlEndpointFor builds the endpoint dialing a node's control channel, -// routing by its id. -func controlEndpointFor(n registry.Node) endpoint { - if ctrl := n.Metadata()[registry.MetaControl]; ctrl != "" { - return endpoint{node: n.ID(), addr: ctrl, target: n.ID()} - } - return endpoint{node: n.ID(), err: "node advertises no control endpoint"} -} - // invocationsFromMetadata reads the invocation names a service advertises in // registry metadata, without dialing. func invocationsFromMetadata(reg registry.Registry, svcName string) ([]*adminpb.InvocationSpec, error) { @@ -298,7 +140,13 @@ func specsToAdmin(in []*controlpb.InvocationSpec) []*adminpb.InvocationSpec { for _, a := range s.Args { args = append(args, &adminpb.ArgSpec{Name: a.Name, Description: a.Description, Required: a.Required}) } - out = append(out, &adminpb.InvocationSpec{Name: s.Name, Description: s.Description, Args: args, Kind: s.Kind, Streaming: s.Streaming}) + out = append(out, &adminpb.InvocationSpec{ + Name: s.Name, + Description: s.Description, + Args: args, + Kind: s.Kind, + Streaming: s.Streaming, + }) } return out } diff --git a/internal/grpc/services/admin/jobs.go b/internal/grpc/services/admin/jobs.go index f48d60a409f..1c2235c9024 100644 --- a/internal/grpc/services/admin/jobs.go +++ b/internal/grpc/services/admin/jobs.go @@ -26,6 +26,7 @@ import ( "github.com/cs3org/reva/v3/pkg/admin" "github.com/cs3org/reva/v3/pkg/admin/adminpb" + "github.com/cs3org/reva/v3/pkg/invoke/client" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -39,12 +40,12 @@ const jobsService = "jobs" // (shared store), since the store — not the ingress runner — decides execution. // jobsEndpoints resolves every live jobs runner. -func (s *svc) jobsEndpoints() ([]endpoint, error) { +func (s *svc) jobsEndpoints() ([]client.Endpoint, error) { reg, err := s.registryHandle() if err != nil { return nil, err } - _, eps, err := resolveSelector(reg, jobsService) + _, eps, err := client.Resolve(reg, jobsService) if err != nil || len(eps) == 0 { return nil, status.Error(codes.NotFound, "admin: no jobs runner is live in the fleet") } @@ -57,11 +58,11 @@ func (s *svc) invokeJobsOne(ctx context.Context, invocation string, args map[str if err != nil { return nil, err } - res := invokeOne(ctx, eps[0], invocation, args) + res := client.InvokeOne(ctx, eps[0], invocation, args) if res.Error != "" { return nil, status.Errorf(codes.Internal, "admin: jobs %s: %s", invocation, res.Error) } - return res, nil + return &adminpb.NodeResult{Node: res.Node, ResultJson: res.ResultJSON, Error: res.Error}, nil } func (s *svc) InspectJobs(ctx context.Context, _ *adminpb.InspectJobsRequest) (*adminpb.InspectJobsResponse, error) { diff --git a/internal/grpc/services/admin/orchestrate_test.go b/internal/grpc/services/admin/orchestrate_test.go index 70bfc90100e..c393ac163ef 100644 --- a/internal/grpc/services/admin/orchestrate_test.go +++ b/internal/grpc/services/admin/orchestrate_test.go @@ -21,6 +21,7 @@ package admin import ( "testing" + "github.com/cs3org/reva/v3/pkg/invoke/client" "github.com/cs3org/reva/v3/pkg/registry" _ "github.com/cs3org/reva/v3/pkg/registry/memory" ) @@ -59,26 +60,26 @@ func testRegistry(t *testing.T) registry.Registry { func TestResolveSelectorByAddress(t *testing.T) { reg := testRegistry(t) - _, eps, err := resolveSelector(reg, "hostA:9001") + _, eps, err := client.Resolve(reg, "hostA:9001") if err != nil { - t.Fatalf("resolveSelector: %v", err) + t.Fatalf("client.Resolve: %v", err) } - if len(eps) != 1 || eps[0].node != "hostA:9001/storageprovider" { + if len(eps) != 1 || eps[0].Node != "hostA:9001/storageprovider" { t.Fatalf("expected the hostA:9001 instance, got %+v", eps) } } func TestResolveSelectorByHost(t *testing.T) { reg := testRegistry(t) - _, eps, err := resolveSelector(reg, "hostA") + _, eps, err := client.Resolve(reg, "hostA") if err != nil { - t.Fatalf("resolveSelector: %v", err) + t.Fatalf("client.Resolve: %v", err) } if len(eps) != 2 { t.Fatalf("expected both hostA instances, got %+v", eps) } // Sorted by node id. - if eps[0].node != "hostA:9001/storageprovider" || eps[1].node != "hostA:9002/userprovider" { + if eps[0].Node != "hostA:9001/storageprovider" || eps[1].Node != "hostA:9002/userprovider" { t.Fatalf("unexpected instances: %+v", eps) } } @@ -93,11 +94,11 @@ func TestResolveSelectorNameWinsOverHost(t *testing.T) { })); err != nil { t.Fatal(err) } - _, eps, err := resolveSelector(reg, "hostA") + _, eps, err := client.Resolve(reg, "hostA") if err != nil { - t.Fatalf("resolveSelector: %v", err) + t.Fatalf("client.Resolve: %v", err) } - if len(eps) != 1 || eps[0].node != "hostB:9100/hostA" { + if len(eps) != 1 || eps[0].Node != "hostB:9100/hostA" { t.Fatalf("expected the service to win over the host, got %+v", eps) } } @@ -106,9 +107,9 @@ func TestResolveSelectorNameWinsOverHost(t *testing.T) { // instance's control endpoint. func TestResolveSelectorByName(t *testing.T) { reg := testRegistry(t) - svc, eps, err := resolveSelector(reg, "storageprovider") + svc, eps, err := client.Resolve(reg, "storageprovider") if err != nil { - t.Fatalf("resolveSelector: %v", err) + t.Fatalf("client.Resolve: %v", err) } if svc != "storageprovider" { t.Errorf("unexpected service: %s", svc) @@ -116,7 +117,7 @@ func TestResolveSelectorByName(t *testing.T) { if len(eps) != 2 { t.Fatalf("expected 2 endpoints, got %d", len(eps)) } - got := map[string]bool{eps[0].addr: true, eps[1].addr: true} + got := map[string]bool{eps[0].Addr: true, eps[1].Addr: true} if !got["hostA:9500"] || !got["hostB:9500"] { t.Errorf("expected both control endpoints, got %+v", eps) } @@ -126,26 +127,26 @@ func TestResolveSelectorByName(t *testing.T) { // still resolves the service name to route with. func TestResolveSelectorByID(t *testing.T) { reg := testRegistry(t) - svc, eps, err := resolveSelector(reg, "hostB:9001/storageprovider") + svc, eps, err := client.Resolve(reg, "hostB:9001/storageprovider") if err != nil { - t.Fatalf("resolveSelector: %v", err) + t.Fatalf("client.Resolve: %v", err) } if svc != "storageprovider" { t.Errorf("unexpected service: %s", svc) } - if len(eps) != 1 || eps[0].node != "hostB:9001/storageprovider" || eps[0].addr != "hostB:9500" { + if len(eps) != 1 || eps[0].Node != "hostB:9001/storageprovider" || eps[0].Addr != "hostB:9500" { t.Fatalf("expected the single hostB instance, got %+v", eps) } // The control channel routes on the node id, so the exact instance is reached. - if eps[0].target != "hostB:9001/storageprovider" { - t.Errorf("expected target to be the node id, got %q", eps[0].target) + if eps[0].Target != "hostB:9001/storageprovider" { + t.Errorf("expected target to be the node id, got %q", eps[0].Target) } } // TestResolveSelectorUnknownID rejects a node id that is not registered. func TestResolveSelectorUnknownID(t *testing.T) { reg := testRegistry(t) - if _, _, err := resolveSelector(reg, "hostZ:9999/storageprovider"); err == nil { + if _, _, err := client.Resolve(reg, "hostZ:9999/storageprovider"); err == nil { t.Fatal("expected error for unknown instance id") } } @@ -153,9 +154,9 @@ func TestResolveSelectorUnknownID(t *testing.T) { // TestResolveSelectorFleet checks that "*" resolves every live instance. func TestResolveSelectorFleet(t *testing.T) { reg := testRegistry(t) - _, eps, err := resolveSelector(reg, "*") + _, eps, err := client.Resolve(reg, "*") if err != nil { - t.Fatalf("resolveSelector: %v", err) + t.Fatalf("client.Resolve: %v", err) } if len(eps) != 3 { t.Fatalf("expected all 3 instances, got %+v", eps) @@ -166,7 +167,7 @@ func TestResolveSelectorFleet(t *testing.T) { // cleanly. func TestResolveSelectorNoMatch(t *testing.T) { reg := testRegistry(t) - if _, _, err := resolveSelector(reg, "nope"); err == nil { + if _, _, err := client.Resolve(reg, "nope"); err == nil { t.Fatal("expected error for a selector matching nothing") } } diff --git a/internal/grpc/services/admin/peer.go b/internal/grpc/services/admin/peer.go deleted file mode 100644 index 98f0f2bafa3..00000000000 --- a/internal/grpc/services/admin/peer.go +++ /dev/null @@ -1,57 +0,0 @@ -// Copyright 2018-2026 CERN -// -// 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. -// -// In applying this license, CERN does not waive the privileges and immunities -// granted to it by virtue of its status as an Intergovernmental Organization -// or submit itself to any jurisdiction. - -package admin - -import ( - "sync" - - "github.com/cs3org/reva/v3/pkg/control/controlpb" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" -) - -// peerConns caches gRPC connections to peer control endpoints; the caller's -// admin token rides the outgoing context. -var peerConns = struct { - mu sync.Mutex - conns map[string]*grpc.ClientConn -}{conns: map[string]*grpc.ClientConn{}} - -func peerConn(address string) (*grpc.ClientConn, error) { - peerConns.mu.Lock() - defer peerConns.mu.Unlock() - if c, ok := peerConns.conns[address]; ok { - return c, nil - } - c, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials())) - if err != nil { - return nil, err - } - peerConns.conns[address] = c - return c, nil -} - -// controlClientAt returns a Control client for a peer control endpoint address. -func controlClientAt(address string) (controlpb.ControlClient, error) { - conn, err := peerConn(address) - if err != nil { - return nil, err - } - return controlpb.NewControlClient(conn), nil -} diff --git a/internal/grpc/services/admin/read.go b/internal/grpc/services/admin/read.go index 94d4c29066c..c6f89a896c5 100644 --- a/internal/grpc/services/admin/read.go +++ b/internal/grpc/services/admin/read.go @@ -27,6 +27,7 @@ import ( "github.com/cs3org/reva/v3/pkg/admin/adminpb" "github.com/cs3org/reva/v3/pkg/invoke" + "github.com/cs3org/reva/v3/pkg/invoke/client" "github.com/cs3org/reva/v3/pkg/registry" "github.com/cs3org/reva/v3/pkg/service" "google.golang.org/grpc/codes" @@ -140,7 +141,7 @@ func (s *svc) GetServiceConfig(ctx context.Context, req *adminpb.GetServiceConfi if err != nil { return nil, err } - _, eps, err := resolveSelector(reg, req.Service) + _, eps, err := client.Resolve(reg, req.Service) if err != nil { return nil, status.Errorf(codes.NotFound, "admin: %v", err) } diff --git a/internal/grpc/services/admin/stream.go b/internal/grpc/services/admin/stream.go index c09c477e870..9a0797fe702 100644 --- a/internal/grpc/services/admin/stream.go +++ b/internal/grpc/services/admin/stream.go @@ -27,6 +27,7 @@ import ( "github.com/cs3org/reva/v3/pkg/admin" "github.com/cs3org/reva/v3/pkg/admin/adminpb" "github.com/cs3org/reva/v3/pkg/control/controlpb" + "github.com/cs3org/reva/v3/pkg/invoke/client" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -44,7 +45,7 @@ func (s *svc) InvokeStream(req *adminpb.InvokeRequest, stream adminpb.AdminAPI_I if err != nil { return err } - _, eps, err := resolveSelector(reg, req.Service) + _, eps, err := client.Resolve(reg, req.Service) if err != nil { return status.Errorf(codes.NotFound, "admin: resolving %q: %v", req.Service, err) } @@ -58,7 +59,7 @@ func (s *svc) InvokeStream(req *adminpb.InvokeRequest, stream adminpb.AdminAPI_I var wg sync.WaitGroup for _, ep := range eps { wg.Add(1) - go func(ep endpoint) { + go func(ep client.Endpoint) { defer wg.Done() streamUpstream(ctx, ep, req.Invocation, req.Args, items) }(ep) @@ -83,7 +84,7 @@ func (s *svc) InvokeStream(req *adminpb.InvokeRequest, stream adminpb.AdminAPI_I // streamUpstream forwards one endpoint's stream, node-labelled, into items. An // unreachable endpoint yields a single per-node error item rather than failing // the whole fan-in. -func streamUpstream(ctx context.Context, ep endpoint, invocation string, args map[string]string, items chan<- *adminpb.InvokeStreamResponse) { +func streamUpstream(ctx context.Context, ep client.Endpoint, invocation string, args map[string]string, items chan<- *adminpb.InvokeStreamResponse) { send := func(it *adminpb.InvokeStreamResponse) bool { select { case items <- it: @@ -92,18 +93,18 @@ func streamUpstream(ctx context.Context, ep endpoint, invocation string, args ma return false } } - if ep.err != "" { - send(&adminpb.InvokeStreamResponse{Node: ep.node, Error: ep.err}) + if ep.Err != "" { + send(&adminpb.InvokeStreamResponse{Node: ep.Node, Error: ep.Err}) return } - cli, err := controlClientAt(ep.addr) + cli, err := client.ControlClientAt(ep.Addr) if err != nil { - send(&adminpb.InvokeStreamResponse{Node: ep.node, Error: err.Error()}) + send(&adminpb.InvokeStreamResponse{Node: ep.Node, Error: err.Error()}) return } - up, err := cli.InvokeStream(ctx, &controlpb.InvokeRequest{Target: ep.target, Invocation: invocation, Args: args}) + up, err := cli.InvokeStream(ctx, &controlpb.InvokeRequest{Target: ep.Target, Invocation: invocation, Args: args}) if err != nil { - send(&adminpb.InvokeStreamResponse{Node: ep.node, Error: err.Error()}) + send(&adminpb.InvokeStreamResponse{Node: ep.Node, Error: err.Error()}) return } for { @@ -114,11 +115,11 @@ func streamUpstream(ctx context.Context, ep endpoint, invocation string, args ma if err != nil { // A cancelled client is a clean stop, not an error worth surfacing. if ctx.Err() == nil { - send(&adminpb.InvokeStreamResponse{Node: ep.node, Error: err.Error()}) + send(&adminpb.InvokeStreamResponse{Node: ep.Node, Error: err.Error()}) } return } - if !send(&adminpb.InvokeStreamResponse{Node: ep.node, ResultJson: msg.ResultJson, Error: msg.Error}) { + if !send(&adminpb.InvokeStreamResponse{Node: ep.Node, ResultJson: msg.ResultJson, Error: msg.Error}) { return } } diff --git a/pkg/invoke/client/client.go b/pkg/invoke/client/client.go new file mode 100644 index 00000000000..9a2b6474a9a --- /dev/null +++ b/pkg/invoke/client/client.go @@ -0,0 +1,250 @@ +// Copyright 2018-2026 CERN +// +// 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. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +// Package client resolves invocation targets through the service registry +// and runs invocations over the control channel. It is the shared +// machinery under the Admin API's Invoke and any other fleet-internal +// consumer of invocations (e.g. the stats prometheus collector). +package client + +import ( + "context" + "fmt" + "net" + "sort" + "strings" + "sync" + "time" + + "github.com/cs3org/reva/v3/pkg/control/controlpb" + "github.com/cs3org/reva/v3/pkg/registry" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// Endpoint is one resolved invocation target: the node id to report, the +// control address to dial, and the target the control channel routes on. +// Err set means the target could not be resolved. +type Endpoint struct { + Node string + Addr string + Target string + Err string +} + +// NodeResult is the outcome of one peer invocation. +type NodeResult struct { + Node string + ResultJSON string + Error string +} + +// PerNodeTimeout bounds a single peer invocation so one slow or offline +// node never stalls a fleet-wide fan-out. +const PerNodeTimeout = 10 * time.Second + +// Resolve maps a selector to its control endpoints: a node id +// "host:port/service" targets one instance, a service name every live one, a +// partial id ("host:port" or a bare host) every instance at that address or on +// that machine, and "*" every live instance in the fleet. It returns the +// resolved service name when the selector was one. +func Resolve(reg registry.Registry, selector string) (string, []Endpoint, error) { + // "*": every live instance in the fleet. + if selector == "*" { + eps := endpointsMatching(reg, func(registry.Node) bool { return true }) + if len(eps) == 0 { + return "", nil, fmt.Errorf("no live instances in the fleet") + } + return selector, eps, nil + } + + // Node id "host:port/service": one exact instance. + if i := strings.LastIndex(selector, "/"); i >= 0 { + svcName := selector[i+1:] + if svcName == "" { + return "", nil, fmt.Errorf("invalid instance id %q", selector) + } + sv, err := reg.GetService(svcName) + if err != nil { + return "", nil, fmt.Errorf("instance %q: service %q not found", selector, svcName) + } + for _, n := range sv.Nodes() { + if n.ID() == selector { + return svcName, []Endpoint{EndpointFor(n)}, nil + } + } + return "", nil, fmt.Errorf("instance %q not found", selector) + } + + // Plain service name: every live instance. + if sv, err := reg.GetService(selector); err == nil && len(sv.Nodes()) > 0 { + var eps []Endpoint + for _, n := range sv.Nodes() { + // A drained node is out of service rotation but still alive and + // control-reachable — keep it so it can be enabled again (and so + // logs/stack/config still work against it). Only offline is skipped. + if nodeState(n) == registry.StateOffline { + continue + } + eps = append(eps, EndpointFor(n)) + } + if len(eps) == 0 { + return "", nil, fmt.Errorf("service %q has no live instances", selector) + } + return selector, eps, nil + } + + // Partial id: "host:port" targets every instance at that address, a bare + // host every instance on that machine. + if eps := endpointsMatchingAddress(reg, selector); len(eps) > 0 { + return selector, eps, nil + } + + return "", nil, fmt.Errorf("%q matches no service, instance, address or host", selector) +} + +// FanOut invokes every endpoint in parallel. An unreachable node is a +// per-node error rather than a failure of the whole call. +func FanOut(ctx context.Context, eps []Endpoint, invocation string, args map[string]string) []NodeResult { + results := make([]NodeResult, len(eps)) + var wg sync.WaitGroup + for i, ep := range eps { + wg.Add(1) + go func(i int, ep Endpoint) { + defer wg.Done() + results[i] = InvokeOne(ctx, ep, invocation, args) + }(i, ep) + } + wg.Wait() + return results +} + +// InvokeOne runs a single peer invocation with a bounded timeout. +func InvokeOne(ctx context.Context, ep Endpoint, invocation string, args map[string]string) NodeResult { + if ep.Err != "" { + return NodeResult{Node: ep.Node, Error: ep.Err} + } + cli, err := ControlClientAt(ep.Addr) + if err != nil { + return NodeResult{Node: ep.Node, Error: err.Error()} + } + cctx, cancel := context.WithTimeout(ctx, PerNodeTimeout) + defer cancel() + resp, err := cli.Invoke(cctx, &controlpb.InvokeRequest{Target: ep.Target, Invocation: invocation, Args: args}) + if err != nil { + return NodeResult{Node: ep.Node, Error: err.Error()} + } + return NodeResult{Node: ep.Node, ResultJSON: resp.ResultJson, Error: resp.Error} +} + +// EndpointFor builds the endpoint dialing a node's control channel, +// routing by its id. +func EndpointFor(n registry.Node) Endpoint { + if ctrl := n.Metadata()[registry.MetaControl]; ctrl != "" { + return Endpoint{Node: n.ID(), Addr: ctrl, Target: n.ID()} + } + return Endpoint{Node: n.ID(), Err: "node advertises no control endpoint"} +} + +// endpointsMatchingAddress resolves a partial node id: "host:port" matches the +// live instances bound to that address, a bare host those on that machine (by +// the id's host part or the node's host metadata). +func endpointsMatchingAddress(reg registry.Registry, selector string) []Endpoint { + byAddress := strings.Contains(selector, ":") + return endpointsMatching(reg, func(n registry.Node) bool { + if byAddress { + return strings.HasPrefix(n.ID(), selector+"/") + } + return onHost(n, selector) + }) +} + +// endpointsMatching gathers the live instances accepted by match, sorted by +// node id. +func endpointsMatching(reg registry.Registry, match func(registry.Node) bool) []Endpoint { + svcs, err := reg.ListServices() + if err != nil { + return nil + } + var eps []Endpoint + for _, sv := range svcs { + for _, n := range sv.Nodes() { + // Drained nodes stay reachable for control (see Resolve); only + // offline is skipped. + if nodeState(n) == registry.StateOffline { + continue + } + if !match(n) { + continue + } + eps = append(eps, EndpointFor(n)) + } + } + sort.Slice(eps, func(i, j int) bool { return eps[i].Node < eps[j].Node }) + return eps +} + +// onHost reports whether a node runs on the given host, by the host part of its +// id's address or by its host metadata. +func onHost(n registry.Node, host string) bool { + id := n.ID() + if i := strings.LastIndex(id, "/"); i >= 0 { + if h, _, err := net.SplitHostPort(id[:i]); err == nil && h == host { + return true + } + } + return n.Metadata()["host"] == host +} + +// nodeState reads a node's self-reported state, defaulting to ready. +func nodeState(n registry.Node) string { + if st := n.Metadata()[registry.MetaState]; st != "" { + return st + } + return registry.StateReady +} + +// peerConns pools one client connection per control address. +var peerConns = struct { + mu sync.Mutex + conns map[string]*grpc.ClientConn +}{conns: map[string]*grpc.ClientConn{}} + +func peerConn(address string) (*grpc.ClientConn, error) { + peerConns.mu.Lock() + defer peerConns.mu.Unlock() + if c, ok := peerConns.conns[address]; ok { + return c, nil + } + c, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return nil, err + } + peerConns.conns[address] = c + return c, nil +} + +// ControlClientAt returns a Control client for a peer control endpoint +// address, pooling the underlying connection. +func ControlClientAt(address string) (controlpb.ControlClient, error) { + conn, err := peerConn(address) + if err != nil { + return nil, err + } + return controlpb.NewControlClient(conn), nil +} From 1905e54e4b775be020b6eccbe62d1ce1640d63c6 Mon Sep 17 00:00:00 2001 From: Gianmaria Del Monte Date: Fri, 28 Aug 2026 16:23:40 +0200 Subject: [PATCH 2/5] Add pkg/stats: self-describing service statistics A driver implements the stats.Reporter capability; its service exposes the payload through a 'stats' invocation. Payloads carry their own metric names, kinds and labels, plus per-owner aggregates. --- pkg/stats/stats.go | 199 ++++++++++++++++++++++++++++++++++++++++ pkg/stats/stats_test.go | 111 ++++++++++++++++++++++ 2 files changed, 310 insertions(+) create mode 100644 pkg/stats/stats.go create mode 100644 pkg/stats/stats_test.go diff --git a/pkg/stats/stats.go b/pkg/stats/stats.go new file mode 100644 index 00000000000..158cecf8e8e --- /dev/null +++ b/pkg/stats/stats.go @@ -0,0 +1,199 @@ +// Copyright 2018-2026 CERN +// +// 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. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +// Package stats defines the self-describing statistics payload a service +// can expose through its `stats` invocation, and the Reporter capability +// a driver implements to produce it. +// +// A service whose manager implements Reporter registers the "stats" +// invocation; the stats prometheus collector (pkg/prom/stats) discovers +// those services through the registry and turns their payloads into +// metrics. The payload fully describes its metrics (names, kinds, labels), +// so new services need no collector changes. +// +// Counters derived from creation-time row counts are monotonic as long as +// rows are soft-deleted; a hard-delete cleanup shows up in Prometheus as a +// counter reset. +package stats + +import ( + "context" + "encoding/json" + "fmt" + "regexp" + + "github.com/cs3org/reva/v3/pkg/invoke" +) + +// Invocation is the conventional invocation name under which services +// expose their statistics. +const Invocation = "stats" + +// Scope declares the state topology of a service's statistics. +const ( + // ScopeShared means every instance reports the same state (e.g. a + // shared database): the collector queries one live instance. + ScopeShared = "shared" + // ScopeInstance means each instance holds its own state: the + // collector fans out and sums. + ScopeInstance = "instance" +) + +// Metric kinds. +const ( + KindGauge = "gauge" + KindCounter = "counter" +) + +// Reporter is the capability a manager (driver) implements to expose +// statistics. Services detect it by type assertion and register the +// "stats" invocation only when it holds. +type Reporter interface { + Stats(ctx context.Context) (*Payload, error) +} + +// Payload is the return value of a "stats" invocation. +type Payload struct { + // Scope is ScopeShared (default when empty) or ScopeInstance. + Scope string `json:"scope,omitempty"` + // Metrics are the self-described metric families. + Metrics []Metric `json:"metrics"` +} + +// Metric is one metric family. Name carries the subject only — the +// collector prefixes the configured namespace and appends the +// Prometheus-conventional `_total` suffix to counters. +type Metric struct { + Name string `json:"name"` + Help string `json:"help,omitempty"` + // Kind is KindGauge or KindCounter. + Kind string `json:"kind"` + // Samples are plain samples. + Samples []Sample `json:"samples,omitempty"` + // OwnerSamples are per-owner aggregates. The collector folds them by + // the owner attributes available at the site (or into plain totals); + // they are never exposed as per-owner series. + OwnerSamples []OwnerSample `json:"owner_samples,omitempty"` +} + +// Sample is one value with its label set. +type Sample struct { + Labels map[string]string `json:"labels,omitempty"` + Value float64 `json:"value"` +} + +// OwnerSample is one per-owner value with an optional label set. +type OwnerSample struct { + Owner string `json:"owner"` + Labels map[string]string `json:"labels,omitempty"` + Value float64 `json:"value"` +} + +var nameRe = regexp.MustCompile(`^[a-z][a-z0-9_]*$`) + +// Validate checks the payload is well formed: legal metric and label +// names, known kinds and scope. +func (p *Payload) Validate() error { + switch p.Scope { + case "", ScopeShared, ScopeInstance: + default: + return fmt.Errorf("stats: unknown scope %q", p.Scope) + } + for _, m := range p.Metrics { + if !nameRe.MatchString(m.Name) { + return fmt.Errorf("stats: metric name %q does not conform to [a-z][a-z0-9_]*", m.Name) + } + if m.Kind != KindGauge && m.Kind != KindCounter { + return fmt.Errorf("stats: metric %s: unknown kind %q", m.Name, m.Kind) + } + for _, s := range m.Samples { + if err := validateLabels(m.Name, s.Labels); err != nil { + return err + } + } + for _, s := range m.OwnerSamples { + if s.Owner == "" { + return fmt.Errorf("stats: metric %s: owner sample without owner", m.Name) + } + if err := validateLabels(m.Name, s.Labels); err != nil { + return err + } + } + } + return nil +} + +func validateLabels(metric string, labels map[string]string) error { + for l := range labels { + if !nameRe.MatchString(l) { + return fmt.Errorf("stats: metric %s: label name %q does not conform to [a-z][a-z0-9_]*", metric, l) + } + } + return nil +} + +// Result validates the payload and converts it to an invoke.Result (the +// JSON-friendly map an invocation returns). +func (p *Payload) Result() (invoke.Result, error) { + if err := p.Validate(); err != nil { + return nil, err + } + data, err := json.Marshal(p) + if err != nil { + return nil, err + } + var out invoke.Result + if err := json.Unmarshal(data, &out); err != nil { + return nil, err + } + return out, nil +} + +// FromJSON parses a payload from an invocation's JSON result and +// validates it. +func FromJSON(data []byte) (*Payload, error) { + var p Payload + if err := json.Unmarshal(data, &p); err != nil { + return nil, fmt.Errorf("stats: parsing payload: %w", err) + } + if err := p.Validate(); err != nil { + return nil, err + } + return &p, nil +} + +// NewInvokeSet returns an invocation set exposing the "stats" invocation +// when the manager implements Reporter, and an empty set otherwise. A +// service embeds the returned set to become invokable, so the capability +// of its driver decides whether it advertises statistics. +func NewInvokeSet(manager any) *invoke.Set { + set := invoke.NewSet() + r, ok := manager.(Reporter) + if !ok { + return set + } + set.Add(Invocation, "Service statistics as a self-describing metrics payload."). + Handle(func(ctx context.Context, _ invoke.Args) (invoke.Result, error) { + p, err := r.Stats(ctx) + if err != nil { + return nil, err + } + return p.Result() + }) + return set +} diff --git a/pkg/stats/stats_test.go b/pkg/stats/stats_test.go new file mode 100644 index 00000000000..15bc820be6b --- /dev/null +++ b/pkg/stats/stats_test.go @@ -0,0 +1,111 @@ +// Copyright 2018-2026 CERN +// +// 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. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +package stats + +import ( + "context" + "encoding/json" + "testing" +) + +func validPayload() *Payload { + return &Payload{ + Scope: ScopeShared, + Metrics: []Metric{ + {Name: "shares", Kind: KindGauge, Samples: []Sample{ + {Labels: map[string]string{"status": "active"}, Value: 3}, + }}, + {Name: "projects", Kind: KindGauge, OwnerSamples: []OwnerSample{ + {Owner: "alice", Value: 2}, + }}, + }, + } +} + +func TestValidate(t *testing.T) { + if err := validPayload().Validate(); err != nil { + t.Fatal(err) + } + + bad := []*Payload{ + {Scope: "wat"}, + {Metrics: []Metric{{Name: "Bad-Name", Kind: KindGauge}}}, + {Metrics: []Metric{{Name: "ok", Kind: "histogram"}}}, + {Metrics: []Metric{{Name: "ok", Kind: KindGauge, + Samples: []Sample{{Labels: map[string]string{"Bad-Label": "x"}}}}}}, + {Metrics: []Metric{{Name: "ok", Kind: KindGauge, + OwnerSamples: []OwnerSample{{Owner: "", Value: 1}}}}}, + } + for i, p := range bad { + if err := p.Validate(); err == nil { + t.Errorf("payload %d: expected validation error", i) + } + } +} + +func TestResultRoundTrip(t *testing.T) { + res, err := validPayload().Result() + if err != nil { + t.Fatal(err) + } + data, err := json.Marshal(res) + if err != nil { + t.Fatal(err) + } + p, err := FromJSON(data) + if err != nil { + t.Fatal(err) + } + if len(p.Metrics) != 2 || p.Metrics[0].Name != "shares" || + p.Metrics[0].Samples[0].Value != 3 || + p.Metrics[1].OwnerSamples[0].Owner != "alice" { + t.Errorf("round trip mangled the payload: %+v", p) + } +} + +type fakeReporter struct{ p *Payload } + +func (f fakeReporter) Stats(context.Context) (*Payload, error) { return f.p, nil } + +func TestNewInvokeSet(t *testing.T) { + // non-reporter: empty set + set := NewInvokeSet(struct{}{}) + if got := len(set.Invocations()); got != 0 { + t.Fatalf("non-reporter registered %d invocations", got) + } + + // reporter: stats invocation wired through + set = NewInvokeSet(fakeReporter{p: validPayload()}) + specs := set.Invocations() + if len(specs) != 1 || specs[0].Name != Invocation { + t.Fatalf("invocations = %+v", specs) + } + res, err := set.Invoke(context.Background(), Invocation, nil) + if err != nil { + t.Fatal(err) + } + data, _ := json.Marshal(res) + p, err := FromJSON(data) + if err != nil { + t.Fatal(err) + } + if p.Scope != ScopeShared || len(p.Metrics) != 2 { + t.Errorf("invoked payload = %+v", p) + } +} From 714f5fc2c28902567a49098805a7e5de93f6c09e Mon Sep 17 00:00:00 2001 From: Gianmaria Del Monte Date: Fri, 28 Aug 2026 16:23:40 +0200 Subject: [PATCH 3/5] sql: implement statistics in the share and link managers Aggregate queries counting shares and links by status, permission class, item type and storage instance, plus creation, creator and recipient counts. --- pkg/share/manager/sql/stats.go | 276 ++++++++++++++++++++++++++++ pkg/share/manager/sql/stats_test.go | 142 ++++++++++++++ 2 files changed, 418 insertions(+) create mode 100644 pkg/share/manager/sql/stats.go create mode 100644 pkg/share/manager/sql/stats_test.go diff --git a/pkg/share/manager/sql/stats.go b/pkg/share/manager/sql/stats.go new file mode 100644 index 00000000000..09429108132 --- /dev/null +++ b/pkg/share/manager/sql/stats.go @@ -0,0 +1,276 @@ +// Copyright 2018-2026 CERN +// +// 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. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +package sql + +import ( + "context" + "time" + + "github.com/cs3org/reva/v3/pkg/share/manager/sql/model" + "github.com/cs3org/reva/v3/pkg/stats" + "gorm.io/gorm" +) + +// The SQL share managers implement stats.Reporter: aggregate GROUP BY +// queries over the share tables, so the schema knowledge stays inside the +// driver that owns it. The services detect the capability and expose it +// as the "stats" invocation; the stats prometheus collector turns the +// payloads into metrics. + +// statusCase buckets each row into exactly one status. Priority: deleted +// wins over orphan, orphan over expired, expired over active — so counts +// sum to the total. +const statusCase = `CASE + WHEN deleted_at IS NOT NULL THEN 'deleted' + WHEN orphan OR orphaned_at IS NOT NULL THEN 'orphan' + WHEN expiration IS NOT NULL AND expiration < ? THEN 'expired' + ELSE 'active' +END` + +// permissionClass maps the permission bitmask (1 read, 2 update, 4 create, +// 8 delete, 16 share) to a small vocabulary: read-only, read-write, +// upload-only (write bits without read), none. +func permissionClass(p uint8) string { + const read, write = 1, 2 | 4 | 8 + switch { + case p&read != 0 && p&write != 0: + return "read-write" + case p&read != 0: + return "read" + case p&write != 0: + return "upload-only" + default: + return "none" + } +} + +// shareCountRow is one aggregate bucket of the by-dimensions query. +type shareCountRow struct { + Status string + ItemType string + Instance string + Permissions uint8 + Count int64 +} + +// countsByDimensions aggregates a share table into (status, item_type, +// instance, permission) buckets, mapping the permission bitmask in Go. +func countsByDimensions(db *gorm.DB, tbl any, kind string) (stats.Metric, error) { + var rows []shareCountRow + err := db.Model(tbl).Unscoped(). + Select(statusCase+" AS status, item_type, instance, permissions, COUNT(*) AS count", time.Now()). + Group("status, item_type, instance, permissions"). + Find(&rows).Error + if err != nil { + return stats.Metric{}, err + } + + // fold the permission bitmask into its class + folded := map[[4]string]int64{} + for _, r := range rows { + key := [4]string{r.Status, r.ItemType, r.Instance, permissionClass(r.Permissions)} + folded[key] += r.Count + } + m := stats.Metric{ + Name: "shares", + Help: "Shares by kind, status, item type, storage instance and permission class.", + Kind: stats.KindGauge, + } + for key, count := range folded { + m.Samples = append(m.Samples, stats.Sample{ + Labels: map[string]string{ + "kind": kind, + "status": key[0], + "item_type": key[1], + "instance": key[2], + "permission": key[3], + }, + Value: float64(count), + }) + } + return m, nil +} + +// createdTotal counts every row ever created (soft deletes keep rows, so +// the count is monotonic and usable as a Prometheus counter). +func createdTotal(db *gorm.DB, tbl any, kind string) (stats.Metric, error) { + var count int64 + if err := db.Model(tbl).Unscoped().Count(&count).Error; err != nil { + return stats.Metric{}, err + } + return stats.Metric{ + Name: "shares_created", + Help: "Shares ever created.", + Kind: stats.KindCounter, + Samples: []stats.Sample{ + {Labels: map[string]string{"kind": kind}, Value: float64(count)}, + }, + }, nil +} + +// creators counts the distinct initiators of live shares. +func creators(db *gorm.DB, tbl any, kind string) (stats.Metric, error) { + var count int64 + if err := db.Model(tbl).Distinct("uid_initiator").Count(&count).Error; err != nil { + return stats.Metric{}, err + } + return stats.Metric{ + Name: "share_creators", + Help: "Distinct users having created shares.", + Kind: stats.KindGauge, + Samples: []stats.Sample{ + {Labels: map[string]string{"kind": kind}, Value: float64(count)}, + }, + }, nil +} + +// Stats implements stats.Reporter for the user/group share manager. +func (m *ShareMgr) Stats(ctx context.Context) (*stats.Payload, error) { + db := m.db.WithContext(ctx) + p := &stats.Payload{Scope: stats.ScopeShared} + + byDim, err := countsByDimensions(db, &model.Share{}, "share") + if err != nil { + return nil, err + } + created, err := createdTotal(db, &model.Share{}, "share") + if err != nil { + return nil, err + } + creat, err := creators(db, &model.Share{}, "share") + if err != nil { + return nil, err + } + + // distinct recipients, by recipient type + type recipientRow struct { + SharedWithIsGroup bool + Count int64 + } + var recRows []recipientRow + err = db.Model(&model.Share{}). + Select("shared_with_is_group, COUNT(DISTINCT share_with) AS count"). + Group("shared_with_is_group"). + Find(&recRows).Error + if err != nil { + return nil, err + } + recipients := stats.Metric{ + Name: "share_recipients", + Help: "Distinct share recipients, by recipient type.", + Kind: stats.KindGauge, + } + for _, r := range recRows { + rtype := "user" + if r.SharedWithIsGroup { + rtype = "group" + } + recipients.Samples = append(recipients.Samples, stats.Sample{ + Labels: map[string]string{"recipient_type": rtype}, + Value: float64(r.Count), + }) + } + + // shares per recipient: max and average + type distRow struct { + Max float64 + Avg float64 + } + var dist distRow + sub := db.Model(&model.Share{}).Select("COUNT(*) AS c").Group("share_with") + err = db.Table("(?) AS per_recipient", sub). + Select("MAX(c) AS max, AVG(c) AS avg"). + Find(&dist).Error + if err != nil { + return nil, err + } + perRecipientMax := stats.Metric{ + Name: "shares_per_recipient_max", + Help: "Maximum number of shares received by a single recipient.", + Kind: stats.KindGauge, + Samples: []stats.Sample{{Value: dist.Max}}, + } + perRecipientAvg := stats.Metric{ + Name: "shares_per_recipient_avg", + Help: "Average number of shares received per recipient.", + Kind: stats.KindGauge, + Samples: []stats.Sample{{Value: dist.Avg}}, + } + + p.Metrics = append(p.Metrics, byDim, created, creat, recipients, perRecipientMax, perRecipientAvg) + return p, nil +} + +// Stats implements stats.Reporter for the public link manager. +func (m *PublicShareMgr) Stats(ctx context.Context) (*stats.Payload, error) { + db := m.db.WithContext(ctx) + p := &stats.Payload{Scope: stats.ScopeShared} + + byDim, err := countsByDimensions(db, &model.PublicLink{}, "link") + if err != nil { + return nil, err + } + created, err := createdTotal(db, &model.PublicLink{}, "link") + if err != nil { + return nil, err + } + creat, err := creators(db, &model.PublicLink{}, "link") + if err != nil { + return nil, err + } + + // live links by protection and quicklink flag + type linkRow struct { + Protected bool + Quicklink bool + Count int64 + } + var linkRows []linkRow + err = db.Model(&model.PublicLink{}). + Select("password <> '' AS protected, quicklink, COUNT(*) AS count"). + Group("protected, quicklink"). + Find(&linkRows).Error + if err != nil { + return nil, err + } + links := stats.Metric{ + Name: "links", + Help: "Public links by password protection and quicklink flag.", + Kind: stats.KindGauge, + } + for _, r := range linkRows { + links.Samples = append(links.Samples, stats.Sample{ + Labels: map[string]string{ + "password_protected": boolLabel(r.Protected), + "quicklink": boolLabel(r.Quicklink), + }, + Value: float64(r.Count), + }) + } + + p.Metrics = append(p.Metrics, byDim, created, creat, links) + return p, nil +} + +func boolLabel(b bool) string { + if b { + return "true" + } + return "false" +} diff --git a/pkg/share/manager/sql/stats_test.go b/pkg/share/manager/sql/stats_test.go new file mode 100644 index 00000000000..cc3fca614c1 --- /dev/null +++ b/pkg/share/manager/sql/stats_test.go @@ -0,0 +1,142 @@ +// Copyright 2018-2026 CERN +// +// 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. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +package sql + +import ( + "testing" + + "github.com/cs3org/reva/v3/pkg/appctx" + "github.com/cs3org/reva/v3/pkg/stats" +) + +// findSample returns the value of the sample matching the given labels in +// the named metric, and whether it was found. +func findSample(p *stats.Payload, name string, labels map[string]string) (float64, bool) { + for _, m := range p.Metrics { + if m.Name != name { + continue + } + samples: + for _, s := range m.Samples { + for k, v := range labels { + if s.Labels[k] != v { + continue samples + } + } + return s.Value, true + } + } + return 0, false +} + +func TestShareManagerStats(t *testing.T) { + mgr, err, teardown := setupSuiteShares(t) + defer teardown(t) + if err != nil { + t.Fatal(err) + } + + userctx := getUserContext("123456") + user, _ := appctx.ContextGetUser(userctx) + file := getRandomFile(user) + + if _, err := mgr.Share(userctx, file, getUserShareGrant("1000", "file")); err != nil { + t.Fatal(err) + } + if _, err := mgr.Share(userctx, file, getUserShareGrant("1001", "file")); err != nil { + t.Fatal(err) + } + + reporter, ok := mgr.(stats.Reporter) + if !ok { + t.Fatal("sql share manager does not implement stats.Reporter") + } + p, err := reporter.Stats(userctx) + if err != nil { + t.Fatal(err) + } + if err := p.Validate(); err != nil { + t.Fatal(err) + } + if p.Scope != stats.ScopeShared { + t.Errorf("scope = %q, want shared", p.Scope) + } + + if v, ok := findSample(p, "shares", map[string]string{ + "kind": "share", "status": "active", "item_type": "file", + }); !ok || v != 2 { + t.Errorf("active file shares = %v (found=%v), want 2", v, ok) + } + if v, ok := findSample(p, "shares_created", map[string]string{"kind": "share"}); !ok || v != 2 { + t.Errorf("shares_created = %v (found=%v), want 2", v, ok) + } + if v, ok := findSample(p, "share_creators", map[string]string{"kind": "share"}); !ok || v != 1 { + t.Errorf("share_creators = %v (found=%v), want 1", v, ok) + } + if v, ok := findSample(p, "share_recipients", map[string]string{"recipient_type": "user"}); !ok || v != 2 { + t.Errorf("user recipients = %v (found=%v), want 2", v, ok) + } + if v, ok := findSample(p, "shares_per_recipient_max", nil); !ok || v != 1 { + t.Errorf("per recipient max = %v (found=%v), want 1", v, ok) + } +} + +func TestPublicShareManagerStats(t *testing.T) { + mgr, err, teardown := setupSuiteLinks(t) + defer teardown(t) + if err != nil { + t.Fatal(err) + } + + userctx := getUserContext("123456") + user, _ := appctx.ContextGetUser(userctx) + file := getRandomFile(user) + + if _, err := mgr.CreatePublicShare(userctx, nil, file, getTestPublicLinkGrant(""), "no password", false, false, ""); err != nil { + t.Fatal(err) + } + if _, err := mgr.CreatePublicShare(userctx, nil, file, getTestPublicLinkGrant("secret"), "with password", false, false, ""); err != nil { + t.Fatal(err) + } + + reporter, ok := mgr.(stats.Reporter) + if !ok { + t.Fatal("sql public share manager does not implement stats.Reporter") + } + p, err := reporter.Stats(userctx) + if err != nil { + t.Fatal(err) + } + if err := p.Validate(); err != nil { + t.Fatal(err) + } + + if v, ok := findSample(p, "shares", map[string]string{"kind": "link", "status": "active"}); !ok || v < 2 { + t.Errorf("active links = %v (found=%v), want >= 2", v, ok) + } + if v, ok := findSample(p, "links", map[string]string{"password_protected": "true"}); !ok || v != 1 { + t.Errorf("protected links = %v (found=%v), want 1", v, ok) + } + if v, ok := findSample(p, "links", map[string]string{"password_protected": "false"}); !ok || v != 1 { + t.Errorf("open links = %v (found=%v), want 1", v, ok) + } + if v, ok := findSample(p, "shares_created", map[string]string{"kind": "link"}); !ok || v != 2 { + t.Errorf("links created = %v (found=%v), want 2", v, ok) + } +} From 70d67a6fa8fac3f232f847b9e17e0d6758514054 Mon Sep 17 00:00:00 2001 From: Gianmaria Del Monte Date: Fri, 28 Aug 2026 16:23:40 +0200 Subject: [PATCH 4/5] Expose the stats invocation from the share providers --- .../grpc/services/publicshareprovider/publicshareprovider.go | 4 ++++ internal/grpc/services/usershareprovider/usershareprovider.go | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/internal/grpc/services/publicshareprovider/publicshareprovider.go b/internal/grpc/services/publicshareprovider/publicshareprovider.go index eb5992ca83c..8589ee92a32 100644 --- a/internal/grpc/services/publicshareprovider/publicshareprovider.go +++ b/internal/grpc/services/publicshareprovider/publicshareprovider.go @@ -27,11 +27,13 @@ import ( provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" "github.com/cs3org/reva/v3/pkg/appctx" "github.com/cs3org/reva/v3/pkg/errtypes" + "github.com/cs3org/reva/v3/pkg/invoke" "github.com/cs3org/reva/v3/pkg/plugin" "github.com/cs3org/reva/v3/pkg/publicshare" "github.com/cs3org/reva/v3/pkg/publicshare/manager/registry" "github.com/cs3org/reva/v3/pkg/rgrpc" "github.com/cs3org/reva/v3/pkg/rgrpc/status" + "github.com/cs3org/reva/v3/pkg/stats" "github.com/cs3org/reva/v3/pkg/utils" "github.com/cs3org/reva/v3/pkg/utils/cfg" "google.golang.org/grpc" @@ -59,6 +61,7 @@ func (c *config) ApplyDefaults() { } type service struct { + *invoke.Set conf *config sm publicshare.Manager allowedPathsForShares []*regexp.Regexp @@ -106,6 +109,7 @@ func New(ctx context.Context, m map[string]any) (rgrpc.Service, error) { } service := &service{ + Set: stats.NewInvokeSet(sm), conf: &c, sm: sm, allowedPathsForShares: allowedPathsForShares, diff --git a/internal/grpc/services/usershareprovider/usershareprovider.go b/internal/grpc/services/usershareprovider/usershareprovider.go index 4c1cb6c247c..cfdafdeee43 100644 --- a/internal/grpc/services/usershareprovider/usershareprovider.go +++ b/internal/grpc/services/usershareprovider/usershareprovider.go @@ -27,11 +27,13 @@ import ( provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" "github.com/cs3org/reva/v3/pkg/appctx" "github.com/cs3org/reva/v3/pkg/errtypes" + "github.com/cs3org/reva/v3/pkg/invoke" "github.com/cs3org/reva/v3/pkg/plugin" "github.com/cs3org/reva/v3/pkg/rgrpc" "github.com/cs3org/reva/v3/pkg/rgrpc/status" "github.com/cs3org/reva/v3/pkg/share" "github.com/cs3org/reva/v3/pkg/share/manager/registry" + "github.com/cs3org/reva/v3/pkg/stats" "github.com/cs3org/reva/v3/pkg/utils" "github.com/cs3org/reva/v3/pkg/utils/cfg" "google.golang.org/grpc" @@ -59,6 +61,7 @@ func (c *config) ApplyDefaults() { } type service struct { + *invoke.Set conf *config sm share.Manager allowedPathsForShares []*regexp.Regexp @@ -106,6 +109,7 @@ func New(ctx context.Context, m map[string]any) (rgrpc.Service, error) { } service := &service{ + Set: stats.NewInvokeSet(sm), conf: &c, sm: sm, allowedPathsForShares: allowedPathsForShares, From 60cf50f0509f73b2f42ccef6aaf8fa153f8897a3 Mon Sep 17 00:00:00 2001 From: Gianmaria Del Monte Date: Fri, 28 Aug 2026 16:23:40 +0200 Subject: [PATCH 5/5] Add the stats prometheus collector Zero configuration: it discovers the services advertising a 'stats' invocation through the registry, queries them over the control channel with its own short-lived admin-scoped token, and serves cached metrics. Per-owner aggregates are folded by site-defined owner attributes read from /etc/revad/owner-attributes.json. --- changelog/unreleased/stats-metrics.md | 25 ++ pkg/prom/loader/loader.go | 1 + pkg/prom/stats/aggregate.go | 178 +++++++++++++ pkg/prom/stats/stats.go | 358 ++++++++++++++++++++++++++ pkg/prom/stats/stats_test.go | 262 +++++++++++++++++++ 5 files changed, 824 insertions(+) create mode 100644 changelog/unreleased/stats-metrics.md create mode 100644 pkg/prom/stats/aggregate.go create mode 100644 pkg/prom/stats/stats.go create mode 100644 pkg/prom/stats/stats_test.go diff --git a/changelog/unreleased/stats-metrics.md b/changelog/unreleased/stats-metrics.md new file mode 100644 index 00000000000..7cbc751f2b3 --- /dev/null +++ b/changelog/unreleased/stats-metrics.md @@ -0,0 +1,25 @@ +Enhancement: Expose service statistics as Prometheus metrics + +Services can now publish statistics (share and public-link counts by +status, permission class, item type and storage instance; creation +counters; creator and recipient counts) on reva's Prometheus endpoint. + +A driver opts in by implementing the `stats.Reporter` capability +(`pkg/stats`), which its service exposes as a `stats` invocation — the +same numbers are then available to operators via +`reva admin invoke stats`. A new zero-configuration collector +served by the existing `prometheus` HTTP service discovers the services +advertising the invocation through the service registry, queries them +over the control channel on a background refresh loop, and serves cached +metrics. The SQL share and public-link managers implement the capability +with in-driver aggregate queries. + +Payloads are self-describing (names, kinds, labels), so new services need +no collector changes. Per-owner aggregates can be enriched with +site-defined owner attributes from a JSON file +(`/etc/revad/owner-attributes.json` by convention): the attribute keys +become metric labels, and reva itself stays agnostic of their meaning. +The invocation fan-out machinery moves from the admin service to the +shared `pkg/invoke/client`. + +https://github.com/cs3org/reva/pull/5792 diff --git a/pkg/prom/loader/loader.go b/pkg/prom/loader/loader.go index 77a04f5597e..108703df60a 100644 --- a/pkg/prom/loader/loader.go +++ b/pkg/prom/loader/loader.go @@ -23,5 +23,6 @@ import ( _ "github.com/cs3org/reva/v3/internal/grpc/interceptors/metrics" _ "github.com/cs3org/reva/v3/internal/http/interceptors/metrics" _ "github.com/cs3org/reva/v3/pkg/prom/base" + _ "github.com/cs3org/reva/v3/pkg/prom/stats" // Add your own here. ) diff --git a/pkg/prom/stats/aggregate.go b/pkg/prom/stats/aggregate.go new file mode 100644 index 00000000000..ff850c9bf04 --- /dev/null +++ b/pkg/prom/stats/aggregate.go @@ -0,0 +1,178 @@ +// Copyright 2018-2026 CERN +// +// 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. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +package stats + +import ( + "encoding/json" + "fmt" + "maps" + "sort" + + "github.com/cs3org/reva/v3/pkg/stats" + "github.com/prometheus/client_golang/prometheus" +) + +// aggregator merges the payloads of a refresh into prometheus const +// metrics. Families with the same name are merged across services and +// payloads (e.g. "shares" from the user share and public link providers); +// identically-labelled samples sum (the per-instance fan-out case). +type aggregator struct { + namespace string + order []string + families map[string]*family + errs []error +} + +type family struct { + name string // rendered, namespace-prefixed, _total-suffixed + help string + kind string + // values sums samples by their serialized label set. + values map[string]float64 + // labelSets remembers the labels behind each key. + labelSets map[string]map[string]string +} + +func newAggregator(namespace string) *aggregator { + return &aggregator{namespace: namespace, families: map[string]*family{}} +} + +// add folds one payload in. Owner samples are enriched with the owner's +// attributes (attrKeys is the label-name union; unmapped owners get +// "unknown") and never emitted per owner. +func (a *aggregator) add(svcName string, p *stats.Payload, attrs map[string]map[string]string, attrKeys []string) { + for _, m := range p.Metrics { + name := a.namespace + "_" + m.Name + if m.Kind == stats.KindCounter { + name += "_total" + } + f, ok := a.families[name] + if !ok { + f = &family{name: name, help: m.Help, kind: m.Kind, + values: map[string]float64{}, labelSets: map[string]map[string]string{}} + a.families[name] = f + a.order = append(a.order, name) + } + if f.kind != m.Kind { + a.errs = append(a.errs, fmt.Errorf("metric %s from %s: kind %s conflicts with %s", m.Name, svcName, m.Kind, f.kind)) + continue + } + if f.help == "" { + f.help = m.Help + } + for _, s := range m.Samples { + f.sum(s.Labels, s.Value) + } + for _, s := range m.OwnerSamples { + f.sum(enrich(s.Labels, s.Owner, attrs, attrKeys), s.Value) + } + } +} + +// enrich merges the owner's attribute labels into the sample labels. +func enrich(labels map[string]string, owner string, attrs map[string]map[string]string, attrKeys []string) map[string]string { + if len(attrKeys) == 0 { + return labels + } + merged := make(map[string]string, len(labels)+len(attrKeys)) + maps.Copy(merged, labels) + ownerAttrs := attrs[owner] + for _, k := range attrKeys { + if v, ok := ownerAttrs[k]; ok && v != "" { + merged[k] = v + } else { + merged[k] = "unknown" + } + } + return merged +} + +// sum accumulates a value under its label set. +func (f *family) sum(labels map[string]string, value float64) { + key := labelKey(labels) + f.values[key] += value + if _, ok := f.labelSets[key]; !ok { + f.labelSets[key] = labels + } +} + +// labelKey serializes a label set deterministically. +func labelKey(labels map[string]string) string { + if len(labels) == 0 { + return "" + } + keys := make([]string, 0, len(labels)) + for k := range labels { + keys = append(keys, k) + } + sort.Strings(keys) + parts := make([][2]string, 0, len(keys)) + for _, k := range keys { + parts = append(parts, [2]string{k, labels[k]}) + } + b, _ := json.Marshal(parts) + return string(b) +} + +// build renders every family into const metrics. Within a family the +// label-name set is the union over its samples; missing labels are "". +func (a *aggregator) build() ([]prometheus.Metric, []error) { + var out []prometheus.Metric + for _, name := range a.order { + f := a.families[name] + + nameSet := map[string]struct{}{} + for _, ls := range f.labelSets { + for k := range ls { + nameSet[k] = struct{}{} + } + } + labelNames := make([]string, 0, len(nameSet)) + for k := range nameSet { + labelNames = append(labelNames, k) + } + sort.Strings(labelNames) + + desc := prometheus.NewDesc(f.name, f.help, labelNames, nil) + valueType := prometheus.GaugeValue + if f.kind == stats.KindCounter { + valueType = prometheus.CounterValue + } + + keys := make([]string, 0, len(f.values)) + for k := range f.values { + keys = append(keys, k) + } + sort.Strings(keys) + for _, key := range keys { + labels := f.labelSets[key] + values := make([]string, len(labelNames)) + for i, ln := range labelNames { + values[i] = labels[ln] + } + m, err := prometheus.NewConstMetric(desc, valueType, f.values[key], values...) + if err != nil { + a.errs = append(a.errs, fmt.Errorf("metric %s: %w", f.name, err)) + continue + } + out = append(out, m) + } + } + return out, a.errs +} diff --git a/pkg/prom/stats/stats.go b/pkg/prom/stats/stats.go new file mode 100644 index 00000000000..55fe5260855 --- /dev/null +++ b/pkg/prom/stats/stats.go @@ -0,0 +1,358 @@ +// Copyright 2018-2026 CERN +// +// 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. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +// Package stats is the prometheus collector turning the fleet's service +// statistics into metrics. It needs no configuration: it discovers the +// services advertising a "stats" invocation through the service registry, +// invokes them over the control channel on a background refresh loop, and +// serves the cached results — a scrape never triggers an invocation. +// +// If an owner-attributes file is present (default +// /etc/revad/owner-attributes.json, a JSON object mapping owner ids to +// attribute key/value pairs), per-owner samples are folded by those +// attributes and the attribute keys become metric labels; without the +// file they collapse into plain totals. The attribute vocabulary is +// entirely the site's: reva only forwards it. +package stats + +import ( + "context" + "encoding/json" + "fmt" + "maps" + "os" + "sort" + "strings" + "sync" + "time" + + userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" + "github.com/cs3org/reva/v3/pkg/appctx" + "github.com/cs3org/reva/v3/pkg/auth/scope" + "github.com/cs3org/reva/v3/pkg/invoke" + "github.com/cs3org/reva/v3/pkg/invoke/client" + "github.com/cs3org/reva/v3/pkg/prom/registry" + svcregistry "github.com/cs3org/reva/v3/pkg/registry" + "github.com/cs3org/reva/v3/pkg/service" + "github.com/cs3org/reva/v3/pkg/stats" + "github.com/cs3org/reva/v3/pkg/token" + jwtmgr "github.com/cs3org/reva/v3/pkg/token/manager/jwt" + "github.com/cs3org/reva/v3/pkg/utils/cfg" + "github.com/prometheus/client_golang/prometheus" + "google.golang.org/grpc/metadata" +) + +func init() { + registry.Register("stats", New) +} + +type config struct { + Stats struct { + // RefreshInterval is how often the fleet is queried (default 5m). + RefreshInterval string `mapstructure:"refresh_interval"` + // Namespace prefixes every metric name (default "reva"). + Namespace string `mapstructure:"namespace"` + // OwnerAttributesFile overrides the conventional attributes path. + OwnerAttributesFile string `mapstructure:"owner_attributes_file"` + } `mapstructure:"stats"` +} + +// DefaultOwnerAttributesFile is the conventional owner-attributes path. +const DefaultOwnerAttributesFile = "/etc/revad/owner-attributes.json" + +// fleet abstracts discovery and invocation for testability. +type fleet interface { + // AuthContext returns a context authorized to call the control + // channel (the collector's own admin-scoped identity). + AuthContext(ctx context.Context) (context.Context, error) + // StatsTargets returns, per service advertising the stats invocation, + // its live control endpoints. + StatsTargets() map[string][]client.Endpoint + // InvokeStats runs the stats invocation on one endpoint. + InvokeStats(ctx context.Context, ep client.Endpoint) (*stats.Payload, error) +} + +// New builds the collector and starts its refresh loop. +func New(ctx context.Context, m map[string]any) ([]prometheus.Collector, error) { + var c config + if err := cfg.Decode(m, &c); err != nil { + return nil, err + } + interval := 5 * time.Minute + if c.Stats.RefreshInterval != "" { + d, err := time.ParseDuration(c.Stats.RefreshInterval) + if err != nil { + return nil, fmt.Errorf("prom stats: refresh_interval: %w", err) + } + interval = d + } + namespace := c.Stats.Namespace + if namespace == "" { + namespace = "reva" + } + attrFile := c.Stats.OwnerAttributesFile + if attrFile == "" { + attrFile = DefaultOwnerAttributesFile + } + + tm, err := jwtmgr.New(nil) + if err != nil { + return nil, fmt.Errorf("prom stats: token manager: %w", err) + } + col := &collector{ + namespace: namespace, + attrFile: attrFile, + fleet: registryFleet{tokenManager: tm}, + errors: map[string]float64{}, + } + go col.run(ctx, interval) + return []prometheus.Collector{col}, nil +} + +// collector caches the fleet's statistics as const metrics. +type collector struct { + namespace string + attrFile string + fleet fleet + + mu sync.Mutex + cached []prometheus.Metric + errors map[string]float64 // per service, cumulative refresh errors + refresh map[string]refreshInfo +} + +type refreshInfo struct { + when time.Time + duration time.Duration +} + +func (c *collector) Describe(ch chan<- *prometheus.Desc) { + prometheus.DescribeByCollect(c, ch) +} + +func (c *collector) Collect(ch chan<- prometheus.Metric) { + c.mu.Lock() + defer c.mu.Unlock() + for _, m := range c.cached { + ch <- m + } + for _, m := range c.selfMetricsLocked() { + ch <- m + } +} + +// run is the refresh loop; the first refresh happens immediately. +func (c *collector) run(ctx context.Context, interval time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + c.refreshOnce(ctx) + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +// refreshOnce queries every discovered stats target and swaps the cache. +func (c *collector) refreshOnce(ctx context.Context) { + log := appctx.GetLogger(ctx) + targets := c.fleet.StatsTargets() + if len(targets) == 0 { + return + } + ctx, err := c.fleet.AuthContext(ctx) + if err != nil { + log.Warn().Err(err).Msg("prom stats: cannot authenticate to the control channel") + return + } + attrs, attrKeys := c.loadOwnerAttributes() + + agg := newAggregator(c.namespace) + c.mu.Lock() + refresh := map[string]refreshInfo{} + maps.Copy(refresh, c.refresh) + errors := c.errors + c.mu.Unlock() + + for svcName, eps := range targets { + start := time.Now() + payloads, err := c.collectService(ctx, eps) + if err != nil { + errors[svcName]++ + log.Warn().Err(err).Str("service", svcName).Msg("prom stats: refresh failed") + continue + } + for _, p := range payloads { + agg.add(svcName, p, attrs, attrKeys) + } + refresh[svcName] = refreshInfo{when: time.Now(), duration: time.Since(start)} + } + + metrics, errs := agg.build() + for _, err := range errs { + log.Warn().Err(err).Msg("prom stats: dropping metric") + } + + c.mu.Lock() + c.cached = metrics + c.refresh = refresh + c.errors = errors + c.mu.Unlock() +} + +// collectService invokes stats on one instance, fanning out to the rest +// only when the payload declares per-instance scope. +func (c *collector) collectService(ctx context.Context, eps []client.Endpoint) ([]*stats.Payload, error) { + if len(eps) == 0 { + return nil, fmt.Errorf("no live instances") + } + first, err := c.fleet.InvokeStats(ctx, eps[0]) + if err != nil { + return nil, err + } + payloads := []*stats.Payload{first} + if first.Scope != stats.ScopeInstance { + return payloads, nil + } + for _, ep := range eps[1:] { + p, err := c.fleet.InvokeStats(ctx, ep) + if err != nil { + return nil, err + } + payloads = append(payloads, p) + } + return payloads, nil +} + +// loadOwnerAttributes reads the attributes file; a missing file means no +// enrichment. Returns the owner map and the sorted union of attribute keys. +func (c *collector) loadOwnerAttributes() (map[string]map[string]string, []string) { + data, err := os.ReadFile(c.attrFile) + if err != nil { + return nil, nil + } + var attrs map[string]map[string]string + if err := json.Unmarshal(data, &attrs); err != nil { + return nil, nil + } + keySet := map[string]struct{}{} + for _, kv := range attrs { + for k := range kv { + keySet[k] = struct{}{} + } + } + keys := make([]string, 0, len(keySet)) + for k := range keySet { + keys = append(keys, k) + } + sort.Strings(keys) + return attrs, keys +} + +// selfMetricsLocked renders the collector's own health metrics. +func (c *collector) selfMetricsLocked() []prometheus.Metric { + var out []prometheus.Metric + tsDesc := prometheus.NewDesc(c.namespace+"_stats_refresh_timestamp_seconds", + "Unix time of the last successful stats refresh, per service.", []string{"service"}, nil) + durDesc := prometheus.NewDesc(c.namespace+"_stats_refresh_duration_seconds", + "Duration of the last successful stats refresh, per service.", []string{"service"}, nil) + errDesc := prometheus.NewDesc(c.namespace+"_stats_refresh_errors_total", + "Failed stats refreshes, per service.", []string{"service"}, nil) + for svcName, info := range c.refresh { + out = append(out, + prometheus.MustNewConstMetric(tsDesc, prometheus.GaugeValue, float64(info.when.Unix()), svcName), + prometheus.MustNewConstMetric(durDesc, prometheus.GaugeValue, info.duration.Seconds(), svcName)) + } + for svcName, n := range c.errors { + out = append(out, prometheus.MustNewConstMetric(errDesc, prometheus.CounterValue, n, svcName)) + } + return out +} + +// registryFleet is the production fleet: the process-wide service +// registry plus the control channel. +type registryFleet struct { + tokenManager token.Manager +} + +// AuthContext mints a short-lived admin-scoped token for the collector's +// own identity — the control channel only accepts admin scope — and puts +// it on the outgoing metadata, exactly like an admin fan-out does. The +// token is signed with the deployment's shared JWT secret, so every +// process in the fleet validates it. +func (f registryFleet) AuthContext(ctx context.Context) (context.Context, error) { + u := &userpb.User{ + Id: &userpb.UserId{OpaqueId: "reva:stats-collector", Type: userpb.UserType_USER_TYPE_APPLICATION}, + Username: "stats-collector", + } + scopes, err := scope.AddAdminScope(nil) + if err != nil { + return nil, err + } + tkn, err := f.tokenManager.MintToken(ctx, u, scopes) + if err != nil { + return nil, err + } + return metadata.AppendToOutgoingContext(ctx, appctx.TokenHeader, tkn), nil +} + +// StatsTargets scans registry metadata for services advertising the stats +// invocation — no dialing involved. +func (registryFleet) StatsTargets() map[string][]client.Endpoint { + reg := service.GlobalRegistry() + if reg == nil { + return nil + } + svcs, err := reg.ListServices() + if err != nil { + return nil + } + targets := map[string][]client.Endpoint{} + for _, sv := range svcs { + if !advertisesStats(sv.Nodes()) { + continue + } + if _, eps, err := client.Resolve(reg, sv.Name()); err == nil && len(eps) > 0 { + targets[sv.Name()] = eps + } + } + return targets +} + +func advertisesStats(nodes []svcregistry.Node) bool { + for _, n := range nodes { + for name := range strings.SplitSeq(n.Metadata()[invoke.MetaInvocations], ",") { + if strings.TrimSpace(name) == stats.Invocation { + return true + } + } + } + return false +} + +// InvokeStats runs the stats invocation on one endpoint and parses the +// payload. +func (registryFleet) InvokeStats(ctx context.Context, ep client.Endpoint) (*stats.Payload, error) { + res := client.InvokeOne(ctx, ep, stats.Invocation, nil) + if res.Error != "" { + return nil, fmt.Errorf("%s: %s", res.Node, res.Error) + } + return stats.FromJSON([]byte(res.ResultJSON)) +} diff --git a/pkg/prom/stats/stats_test.go b/pkg/prom/stats/stats_test.go new file mode 100644 index 00000000000..36b0773d9e1 --- /dev/null +++ b/pkg/prom/stats/stats_test.go @@ -0,0 +1,262 @@ +// Copyright 2018-2026 CERN +// +// 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. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +package stats + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/cs3org/reva/v3/pkg/appctx" + "github.com/cs3org/reva/v3/pkg/auth/scope" + "github.com/cs3org/reva/v3/pkg/invoke/client" + "github.com/cs3org/reva/v3/pkg/stats" + jwtmgr "github.com/cs3org/reva/v3/pkg/token/manager/jwt" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "google.golang.org/grpc/metadata" +) + +// fakeFleet serves canned payloads per endpoint node id. +type fakeFleet struct { + targets map[string][]client.Endpoint + payloads map[string]*stats.Payload +} + +func (f fakeFleet) AuthContext(ctx context.Context) (context.Context, error) { return ctx, nil } +func (f fakeFleet) StatsTargets() map[string][]client.Endpoint { return f.targets } +func (f fakeFleet) InvokeStats(_ context.Context, ep client.Endpoint) (*stats.Payload, error) { + p, ok := f.payloads[ep.Node] + if !ok { + return nil, fmt.Errorf("no payload for %s", ep.Node) + } + return p, nil +} + +// gather renders the collector's metrics as text lines "name{labels} value". +func gather(t *testing.T, c *collector) map[string]float64 { + t.Helper() + ch := make(chan prometheus.Metric, 100) + c.Collect(ch) + close(ch) + out := map[string]float64{} + for m := range ch { + var d dto.Metric + if err := m.Write(&d); err != nil { + t.Fatal(err) + } + var labels []string + for _, l := range d.Label { + labels = append(labels, l.GetName()+"="+l.GetValue()) + } + key := descName(m.Desc().String()) + "{" + strings.Join(labels, ",") + "}" + switch { + case d.Gauge != nil: + out[key] = d.Gauge.GetValue() + case d.Counter != nil: + out[key] = d.Counter.GetValue() + } + } + return out +} + +// descName extracts the fqName from a Desc's String() form. +func descName(s string) string { + const marker = `fqName: "` + i := strings.Index(s, marker) + if i < 0 { + return s + } + s = s[i+len(marker):] + return s[:strings.Index(s, `"`)] +} + +func shared(metrics ...stats.Metric) *stats.Payload { + return &stats.Payload{Scope: stats.ScopeShared, Metrics: metrics} +} + +func TestRefreshMergesFamiliesAcrossServices(t *testing.T) { + f := fakeFleet{ + targets: map[string][]client.Endpoint{ + "usershareprovider": {{Node: "n1", Addr: "a1", Target: "n1"}}, + "publicshareprovider": {{Node: "n2", Addr: "a2", Target: "n2"}}, + }, + payloads: map[string]*stats.Payload{ + "n1": shared(stats.Metric{Name: "shares", Kind: stats.KindGauge, Samples: []stats.Sample{ + {Labels: map[string]string{"kind": "share", "status": "active"}, Value: 5}, + }}), + "n2": shared(stats.Metric{Name: "shares", Kind: stats.KindGauge, Samples: []stats.Sample{ + {Labels: map[string]string{"kind": "link", "status": "active"}, Value: 2}, + }}), + }, + } + c := &collector{namespace: "reva", attrFile: "/nonexistent", fleet: f, errors: map[string]float64{}} + c.refreshOnce(context.Background()) + + got := gather(t, c) + if got["reva_shares{kind=share,status=active}"] != 5 { + t.Errorf("share sample missing: %v", got) + } + if got["reva_shares{kind=link,status=active}"] != 2 { + t.Errorf("link sample missing: %v", got) + } +} + +func TestInstanceScopeSums(t *testing.T) { + f := fakeFleet{ + targets: map[string][]client.Endpoint{ + "gateway": {{Node: "n1", Addr: "a1"}, {Node: "n2", Addr: "a2"}}, + }, + payloads: map[string]*stats.Payload{ + "n1": {Scope: stats.ScopeInstance, Metrics: []stats.Metric{ + {Name: "sessions", Kind: stats.KindGauge, Samples: []stats.Sample{{Value: 3}}}}}, + "n2": {Scope: stats.ScopeInstance, Metrics: []stats.Metric{ + {Name: "sessions", Kind: stats.KindGauge, Samples: []stats.Sample{{Value: 4}}}}}, + }, + } + c := &collector{namespace: "reva", attrFile: "/nonexistent", fleet: f, errors: map[string]float64{}} + c.refreshOnce(context.Background()) + + got := gather(t, c) + if got["reva_sessions{}"] != 7 { + t.Errorf("instance-scope sum = %v, want 7: %v", got["reva_sessions{}"], got) + } +} + +func TestOwnerAttributeEnrichment(t *testing.T) { + attrFile := filepath.Join(t.TempDir(), "owners.json") + os.WriteFile(attrFile, []byte(`{ + "alice": {"department": "IT", "experiment": "ATLAS"}, + "bob": {"department": "EP"} + }`), 0o644) + + f := fakeFleet{ + targets: map[string][]client.Endpoint{ + "projects": {{Node: "n1", Addr: "a1"}}, + }, + payloads: map[string]*stats.Payload{ + "n1": shared(stats.Metric{Name: "projects", Kind: stats.KindGauge, OwnerSamples: []stats.OwnerSample{ + {Owner: "alice", Value: 2}, + {Owner: "bob", Value: 1}, + {Owner: "carol", Value: 4}, // unmapped + }}), + }, + } + c := &collector{namespace: "reva", attrFile: attrFile, fleet: f, errors: map[string]float64{}} + c.refreshOnce(context.Background()) + + got := gather(t, c) + if got["reva_projects{department=IT,experiment=ATLAS}"] != 2 { + t.Errorf("alice bucket: %v", got) + } + if got["reva_projects{department=EP,experiment=unknown}"] != 1 { + t.Errorf("bob bucket (partial attrs): %v", got) + } + if got["reva_projects{department=unknown,experiment=unknown}"] != 4 { + t.Errorf("carol bucket (unmapped): %v", got) + } +} + +func TestOwnerSamplesCollapseWithoutAttributes(t *testing.T) { + f := fakeFleet{ + targets: map[string][]client.Endpoint{ + "projects": {{Node: "n1", Addr: "a1"}}, + }, + payloads: map[string]*stats.Payload{ + "n1": shared(stats.Metric{Name: "projects", Kind: stats.KindGauge, OwnerSamples: []stats.OwnerSample{ + {Owner: "alice", Value: 2}, + {Owner: "bob", Value: 1}, + }}), + }, + } + c := &collector{namespace: "reva", attrFile: "/nonexistent", fleet: f, errors: map[string]float64{}} + c.refreshOnce(context.Background()) + + got := gather(t, c) + if got["reva_projects{}"] != 3 { + t.Errorf("collapsed total = %v, want 3 (no per-owner series): %v", got["reva_projects{}"], got) + } + for k := range got { + if strings.Contains(k, "alice") || strings.Contains(k, "bob") { + t.Errorf("per-owner series leaked: %s", k) + } + } +} + +func TestCounterSuffixAndSelfMetrics(t *testing.T) { + f := fakeFleet{ + targets: map[string][]client.Endpoint{ + "usershareprovider": {{Node: "n1", Addr: "a1"}}, + "broken": {{Node: "nX", Addr: "aX"}}, // no payload -> error + }, + payloads: map[string]*stats.Payload{ + "n1": shared(stats.Metric{Name: "shares_created", Kind: stats.KindCounter, Samples: []stats.Sample{ + {Labels: map[string]string{"kind": "share"}, Value: 41}, + }}), + }, + } + c := &collector{namespace: "reva", attrFile: "/nonexistent", fleet: f, errors: map[string]float64{}} + c.refreshOnce(context.Background()) + + got := gather(t, c) + if got["reva_shares_created_total{kind=share}"] != 41 { + t.Errorf("counter not suffixed/valued: %v", got) + } + if got["reva_stats_refresh_errors_total{service=broken}"] != 1 { + t.Errorf("refresh error not counted: %v", got) + } + found := false + for k := range got { + if strings.HasPrefix(k, "reva_stats_refresh_timestamp_seconds{service=usershareprovider}") { + found = true + } + } + if !found { + t.Errorf("refresh timestamp missing: %v", got) + } +} + +func TestRegistryFleetAuthContext(t *testing.T) { + tm, err := jwtmgr.New(map[string]any{"secret": "test-secret"}) + if err != nil { + t.Fatal(err) + } + f := registryFleet{tokenManager: tm} + ctx, err := f.AuthContext(context.Background()) + if err != nil { + t.Fatal(err) + } + md, ok := metadata.FromOutgoingContext(ctx) + if !ok || len(md.Get(appctx.TokenHeader)) == 0 { + t.Fatal("no access token on the outgoing metadata") + } + u, scopes, err := tm.DismantleToken(ctx, md.Get(appctx.TokenHeader)[0]) + if err != nil { + t.Fatal(err) + } + if u.Username != "stats-collector" { + t.Errorf("token user = %q", u.Username) + } + if !scope.HasAdminScope(scopes) { + t.Error("token lacks the admin scope required by the control channel") + } +}