Skip to content

Feature/stats metrics #5792

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gmgigi96 wants to merge 5 commits into
base: master
Choose a base branch
from
Open

Feature/stats metrics #5792

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
943c235
Extract the invocation fan-out into pkg/invoke/client
gmgigi96 Aug 28, 2026
1905e54
Add pkg/stats: self-describing service statistics
gmgigi96 Aug 28, 2026
714f5fc
sql: implement statistics in the share and link managers
gmgigi96 Aug 28, 2026
70d67a6
Expose the stats invocation from the share providers
gmgigi96 Aug 28, 2026
60cf50f
Add the stats prometheus collector
gmgigi96 Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions changelog/unreleased/stats-metrics.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 <service> 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
8 changes: 5 additions & 3 deletions internal/grpc/services/admin/fanout_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
188 changes: 18 additions & 170 deletions internal/grpc/services/admin/invoke.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
}
}
Expand All @@ -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)
}
Expand All @@ -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) {
Expand Down Expand Up @@ -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
}
9 changes: 5 additions & 4 deletions internal/grpc/services/admin/jobs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand All @@ -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")
}
Expand All @@ -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) {
Expand Down
Loading
Loading