Allow service.Host to return services (#6348)

Signed-off-by: Paschalis Tsilias <[email protected]>

cmd/internal/flowmode/cmd_run.go

@@ -29,7 +29,6 @@ import (
 	"github.com/grafana/agent/pkg/flow/tracing"
 	"github.com/grafana/agent/pkg/usagestats"
 	"github.com/grafana/agent/service"
-	"github.com/grafana/agent/service/cluster"
 	httpservice "github.com/grafana/agent/service/http"
 	"github.com/grafana/agent/service/labelstore"
 	otel_service "github.com/grafana/agent/service/otel"
@@ -254,7 +253,6 @@ func (fr *flowRun) Run(configPath string) error {
 
 	uiService := uiservice.New(uiservice.Options{
 		UIPrefix: fr.uiPrefix,
-		Cluster:  clusterService.Data().(cluster.Cluster),
 	})
 
 	otelService := otel_service.New(l)

pkg/flow/flow_services.go

@@ -26,6 +26,17 @@ func (f *Flow) GetServiceConsumers(serviceName string) []service.Consumer {
 	return consumers
 }
 
+// GetService implements [service.Host]. It looks up a [service.Service] by
+// name.
+func (f *Flow) GetService(name string) (service.Service, bool) {
+	for _, svc := range f.opts.Services {
+		if svc.Definition().Name == name {
+			return svc, true
+		}
+	}
+	return nil, false
+}
+
 func serviceConsumersForGraph(graph *dag.Graph, serviceName string, includePeerServices bool) []service.Consumer {
 	serviceNode, _ := graph.GetByID(serviceName).(*controller.ServiceNode)
 	if serviceNode == nil {

service/http/http_test.go

@@ -217,3 +217,5 @@ func (fakeHost) ListComponents(moduleID string, opts component.InfoOptions) ([]*
 func (fakeHost) GetServiceConsumers(serviceName string) []service.Consumer { return nil }
 
 func (fakeHost) NewController(id string) service.Controller { return nil }
+
+func (fakeHost) GetService(_ string) (service.Service, bool) { return nil, false }

service/remotecfg/remotecfg_test.go

@@ -156,7 +156,8 @@ func (fakeHost) ListComponents(moduleID string, opts component.InfoOptions) ([]*
 	return nil, fmt.Errorf("no such module %q", moduleID)
 }
 
-func (fakeHost) GetServiceConsumers(serviceName string) []service.Consumer { return nil }
+func (fakeHost) GetServiceConsumers(_ string) []service.Consumer { return nil }
+func (fakeHost) GetService(_ string) (service.Service, bool)     { return nil, false }
 
 func (f fakeHost) NewController(id string) service.Controller {
 	logger, _ := logging.New(io.Discard, logging.DefaultOptions)

service/service.go

@@ -51,6 +51,9 @@ type Host interface {
 	// exist.
 	ListComponents(moduleID string, opts component.InfoOptions) ([]*component.Info, error)
 
+	// GetService gets a running service using its name.
+	GetService(name string) (Service, bool)
+
 	// GetServiceConsumers gets the list of services which depend on a service by
 	// name.
 	GetServiceConsumers(serviceName string) []Consumer

service/ui/ui.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/gorilla/mux"
 	"github.com/grafana/agent/service"
-	"github.com/grafana/agent/service/cluster"
 	http_service "github.com/grafana/agent/service/http"
 	"github.com/grafana/agent/web/api"
 	"github.com/grafana/agent/web/ui"
@@ -21,7 +20,6 @@ const ServiceName = "ui"
 // Options are used to configure the UI service. Options are constant for the
 // lifetime of the UI service.
 type Options struct {
-	Cluster  cluster.Cluster
 	UIPrefix string // Path prefix to host the UI at.
 }
 
@@ -75,9 +73,7 @@ func (s *Service) Data() any {
 func (s *Service) ServiceHandler(host service.Host) (base string, handler http.Handler) {
 	r := mux.NewRouter()
 
-	// TODO(rfratto): allow service.Host to return services so we don't have to
-	// pass the clustering service in Options.
-	fa := api.NewFlowAPI(host, s.opts.Cluster)
+	fa := api.NewFlowAPI(host)
 	fa.RegisterRoutes(path.Join(s.opts.UIPrefix, "/api/v0/web"), r)
 	ui.RegisterRoutes(s.opts.UIPrefix, r)
 

web/api/api.go

@@ -11,19 +11,19 @@ import (
 
 	"github.com/gorilla/mux"
 	"github.com/grafana/agent/component"
+	"github.com/grafana/agent/service"
 	"github.com/grafana/agent/service/cluster"
 	"github.com/prometheus/prometheus/util/httputil"
 )
 
 // FlowAPI is a wrapper around the component API.
 type FlowAPI struct {
-	flow    component.Provider
-	cluster cluster.Cluster
+	flow service.Host
 }
 
 // NewFlowAPI instantiates a new Flow API.
-func NewFlowAPI(flow component.Provider, cluster cluster.Cluster) *FlowAPI {
-	return &FlowAPI{flow: flow, cluster: cluster}
+func NewFlowAPI(flow service.Host) *FlowAPI {
+	return &FlowAPI{flow: flow}
 }
 
 // RegisterRoutes registers all the API's routes.
@@ -93,7 +93,12 @@ func (f *FlowAPI) getClusteringPeersHandler() http.HandlerFunc {
 	return func(w http.ResponseWriter, _ *http.Request) {
 		// TODO(@tpaschalis) Detect if clustering is disabled and propagate to
 		// the Typescript code (eg. via the returned status code?).
-		peers := f.cluster.Peers()
+		svc, found := f.flow.GetService(cluster.ServiceName)
+		if !found {
+			http.Error(w, "cluster service not running", http.StatusInternalServerError)
+			return
+		}
+		peers := svc.Data().(cluster.Cluster).Peers()
 		bb, err := json.Marshal(peers)
 		if err != nil {
 			http.Error(w, err.Error(), http.StatusInternalServerError)