chore(graph): disable HTTP or eventhandlers by configuration

internal/eventstest/events_test_helpers.go

@@ -0,0 +1,44 @@
+package eventstest
+
+import (
+	"encoding/json"
+	"reflect"
+
+	"github.com/google/uuid"
+
+	rev "github.com/opencloud-eu/reva/v2/pkg/events"
+	microevents "go-micro.dev/v4/events"
+)
+
+func NewTestBus() TestBus {
+	return TestBus(make(chan rev.Event))
+}
+
+type TestBus chan rev.Event
+
+func (tb TestBus) Consume(_ string, _ ...microevents.ConsumeOption) (<-chan microevents.Event, error) {
+	ch := make(chan microevents.Event)
+	go func() {
+		for ev := range tb {
+			b, _ := json.Marshal(ev.Event)
+			ch <- microevents.Event{
+				Payload: b,
+				Metadata: map[string]string{
+					rev.MetadatakeyEventID:   ev.ID,
+					rev.MetadatakeyEventType: ev.Type,
+				},
+			}
+		}
+	}()
+	return ch, nil
+}
+
+func (tb TestBus) Publish(e any) string {
+	ev := rev.Event{
+		ID:    uuid.New().String(),
+		Type:  reflect.TypeOf(e).String(),
+		Event: e,
+	}
+	tb <- ev
+	return ev.ID
+}

internal/metricstest/metrics_test_helpers.go

@@ -0,0 +1,143 @@
+package metricstest
+
+import (
+	"fmt"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// copied and adapted from Prometheus testutil.ToFloat64(), since we don't import that package
+func collect(c prometheus.Collector) []prometheus.Metric {
+	result := []prometheus.Metric{}
+	ch := make(chan prometheus.Metric)
+	done := make(chan struct{})
+	go func() {
+		for m := range ch {
+			result = append(result, m)
+		}
+		close(done)
+	}()
+	c.Collect(ch)
+	close(ch)
+	<-done
+	return result
+}
+
+func RequireIsNotSet(t require.TestingT, c prometheus.Collector, msgAndArgs ...any) {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+	if !IsNotSet(t, c, msgAndArgs) {
+		t.FailNow()
+	}
+}
+
+func IsNotSet(t assert.TestingT, c prometheus.Collector, msgAndArgs ...any) bool {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+
+	m := collect(c)
+	if len(m) > 0 {
+		return assert.Fail(t, "Metric exists while expected to not exist", msgAndArgs)
+	} else {
+		return true
+	}
+}
+
+func RequireEqual(t require.TestingT, expected float64, c prometheus.Collector, msgAndArgs ...any) {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+	if !Equal(t, expected, c, msgAndArgs) {
+		t.FailNow()
+	}
+}
+
+// copied and adapted from Prometheus testutil.ToFloat64(), since we don't import that package
+func Equal(t assert.TestingT, expected float64, c prometheus.Collector, msgAndArgs ...any) bool {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+
+	m := collect(c)
+	if !assert.Len(t, m, 1, msgAndArgs...) {
+		return false
+	}
+	pb := &dto.Metric{}
+	err := m[0].Write(pb)
+	if !assert.NoError(t, err, msgAndArgs...) {
+		return false
+	}
+	if pb.Gauge != nil {
+		return assert.Equal(t, expected, pb.Gauge.GetValue(), msgAndArgs...)
+	} else if pb.Counter != nil {
+		return assert.Equal(t, expected, pb.Counter.GetValue(), msgAndArgs...)
+	} else if pb.Untyped != nil {
+		return assert.Equal(t, expected, pb.Untyped.GetValue(), msgAndArgs...)
+	} else {
+		return assert.Fail(t, fmt.Sprintf("collected a non-gauge/counter/untyped metric: %s", pb), msgAndArgs...)
+	}
+}
+
+func RequireEqualWithLabels(t require.TestingT, expectedValue float64, expectedLabels map[string]string, c prometheus.Collector, msgAndArgs ...any) {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+	if !EqualWithLabels(t, expectedValue, expectedLabels, c, msgAndArgs) {
+		t.FailNow()
+	}
+}
+
+func EqualWithLabels(t assert.TestingT, expectedValue float64, expectedLabels map[string]string, c prometheus.Collector, msgAndArgs ...any) bool {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+
+	m := collect(c)
+	if !assert.Len(t, m, 1, "collected %d metrics instead of exactly 1", len(m)) {
+		return false
+	}
+	pb := &dto.Metric{}
+	err := m[0].Write(pb)
+	if !assert.NoError(t, err) {
+		return false
+	}
+	if pb.Gauge != nil {
+		if !assert.Equal(t, expectedValue, pb.Gauge.GetValue()) {
+			return false
+		}
+	} else if pb.Counter != nil {
+		if !assert.Equal(t, expectedValue, pb.Counter.GetValue()) {
+			return false
+		}
+	} else if pb.Untyped != nil {
+		if !assert.Equal(t, expectedValue, pb.Untyped.GetValue()) {
+			return false
+		}
+	} else {
+		return assert.Fail(t, "collected a non-gauge/counter/untyped metric: %s", pb)
+	}
+
+	if !assert.NotNil(t, pb.Label) {
+		return false
+	}
+	actualLabels := map[string]string{}
+	for _, label := range pb.Label {
+		if !assert.NotNil(t, label) {
+			return false
+		}
+		if !assert.NotNil(t, label.Name) {
+			return false
+		}
+		if !assert.NotNil(t, label.Value) {
+			return false
+		}
+		actualLabels[*label.Name] = *label.Value
+	}
+	return assert.Equal(t, expectedLabels, actualLabels, msgAndArgs)
+}

services/graph/README.md

@@ -168,7 +168,7 @@ The output of this command includes the following information for each role:
 * `Condition`
 * `Allowed resource actions`
 
-**Example output (shortned)**
+**Example output (shortened)**
 
 ```bash
 +--------------------------------------+----------+--------------------------------+--------------------------------+------------------------------------------+
@@ -184,3 +184,22 @@ The output of this command includes the following information for each role:
 +--------------------------------------+----------+--------------------------------+--------------------------------+------------------------------------------+
 ```
 
+## API Handlers
+
+To specialize `graph` service instances in order to scale them independently, it is possible to disable its API handlers:
+
+* `GRAPH_HTTP_DISABLE`: when set to `true`, the service does not listen on HTTP and only consumes events (defaults to `false`)
+* `GRAPH_EVENTS_DISABLE_CONSUMER`: when set to `true`, the service does not consome events and only listens on HTTP (defaults to `false`)
+
+## Metrics
+
+The `graph` service provides the following metrics:
+
+| Name | Description |
+| ---- | ----------- |
+| `opencloud_graph_build_info{version=...}` | Contains a label `version` that is set to the current version of the service, and always has a value of `1` |
+| `opencloud_graph_events_enabled` | Is set to `1` if the Events API handler is enabled, or `0` if not |
+| `opencloud_graph_http_enabled` | Is set to `1` if the HTTP API handler is enabled, or `0` if not |
+| `opencloud_graph_events{event=...,result=...}` | Counts the number of events that have been consumed, with a `event` label that contains the name of the event, and a `result` label that is set to `success` or `failure` |
+| `opencloud_graph_events_invalid` | Counts the number of invalid events that are malformed or are missing required data |
+| `opencloud_graph_events_unsupported` | Counts the numbef of consumed events that cannot be processes by this service, should always be `0` |

services/graph/pkg/command/server.go

@@ -7,16 +7,23 @@ import (
 	"strings"
 
 	"github.com/opencloud-eu/opencloud/pkg/config/configlog"
+	"github.com/opencloud-eu/opencloud/pkg/generators"
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	natspkg "github.com/opencloud-eu/opencloud/pkg/nats"
 	"github.com/opencloud-eu/opencloud/pkg/runner"
 	"github.com/opencloud-eu/opencloud/pkg/tracing"
 	"github.com/opencloud-eu/opencloud/pkg/version"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/config/parser"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/identity"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/metrics"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/server/debug"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/server/http"
+	evc "github.com/opencloud-eu/opencloud/services/graph/pkg/service/events"
+	svc "github.com/opencloud-eu/opencloud/services/graph/pkg/service/v0"
+	"github.com/opencloud-eu/reva/v2/pkg/events"
+	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
+	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -46,7 +53,14 @@ func Server(cfg *config.Config) *cobra.Command {
 			}
 			ctx := cfg.Context
 
-			mtrcs := metrics.New()
+			prom := prometheus.DefaultRegisterer
+
+			// note that the function we pass here is tasked with decomposing Graph HTTP API
+			// request URL patterns into information that is then used for labels in metrics
+			// to track HTTP request processing durations, and it is located there to be close
+			// to the HTTP API route definitions, to improve chances of adapting it accordingly
+			// whenever those routes should change in the future
+			mtrcs := metrics.New(prom, svc.DecomposeGraphApiRequestPattern)
 			mtrcs.BuildInfo.WithLabelValues(version.GetString()).Set(1)
 
 			var kv jetstream.KeyValue
@@ -78,9 +92,45 @@ func Server(cfg *config.Config) *cobra.Command {
 				}
 			}
 
+			identityBackendName := cfg.Identity.Backend // contains the name of the backend implementation to use
+
+			// since the identity backend in use is of prime importance to understand issues through logs, every
+			// log entry should contain a 'backend' entry with the name of the backend in use from here on:
+			logger = log.Logger{Logger: logger.With().Str("backend", identityBackendName).Logger()}
+
+			identityBackend, eduBackend, err := identity.CreateIdentityBackends(
+				identityBackendName,
+				cfg,
+				&logger,
+				prom,
+				traceProvider,
+			)
+
+			if err != nil {
+				logger.Error().Err(err).Msg("Error initializing the identity backend")
+				return fmt.Errorf("could not initialize identity backend: %w", err)
+			}
+
+			var eventsStream events.Stream
+			if cfg.Events.Endpoint != "" {
+				var err error
+				connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus)
+				eventsStream, err = stream.NatsFromConfig(connName, false, cfg.Events.ToNatsConfig())
+				if err != nil {
+					logger.Error().Err(err).Msg("Error initializing events publisher")
+					return fmt.Errorf("could not initialize events publisher: %w", err)
+				}
+			}
+
 			gr := runner.NewGroup()
-			{
+
+			if !cfg.HTTP.Disabled {
+				mtrcs.HttpEnabled.Set(1)
+
 				server, err := http.Server(
+					identityBackend,
+					eduBackend,
+					eventsStream,
 					http.Logger(logger),
 					http.Context(ctx),
 					http.Config(cfg),
@@ -92,8 +142,37 @@ func Server(cfg *config.Config) *cobra.Command {
 					logger.Error().Err(err).Str("transport", "http").Msg("Failed to initialize server")
 					return err
 				}
-
 				gr.Add(runner.NewGoMicroHttpServerRunner(cfg.Service.Name+".http", server))
+			} else {
+				mtrcs.HttpEnabled.Set(0)
+				logger.Info().Str("transport", "http").Msg("HTTP server is disabled")
+			}
+
+			if !cfg.Events.DisabledConsumer {
+				mtrcs.EventsEnabled.Set(1)
+
+				// even if events are enabled, we still need to differentiate between whether this process
+				// show be consuming events or not (and even when that is disabled, we still need to be
+				// able to produce events), which is why this is a separate setting;
+				// for context, see https://github.com/opencloud-eu/opencloud/issues/1312
+
+				logger := &log.Logger{Logger: logger.With().Str("transport", "events").Logger()}
+				eventConsumer, err := evc.NewService(cfg.Context, eventsStream, identityBackend, mtrcs, logger)
+				if err != nil {
+					return fmt.Errorf("could not initialize events consumer: %w", err)
+				}
+
+				gr.Add(runner.New(cfg.Service.Name+".svc", func() error {
+					return eventConsumer.Start()
+				}, func() {
+					err := eventConsumer.Close()
+					if err != nil {
+						logger.Error().Err(err).Msg("failed to stop event consumer")
+					}
+				}))
+			} else {
+				mtrcs.EventsEnabled.Set(0)
+				logger.Info().Str("transport", "events").Msg("event consumer is disabled")
 			}
 
 			{