chore: adds rpc pings to health

cmd/exporter/main.go

@@ -111,7 +111,7 @@ func main() {
 	logger.Info("bus started")
 
 	// Create the API object
-	server := api.New()
+	server := api.New(api.WithExportPluginSystem(pluginsystem))
 
 	// Scheduler manager
 	scrape := scraper.NewManager(ctx, b)

pkg/api/api.go

@@ -13,6 +13,7 @@ import (
 	"github.com/prometheus/common/version"
 	"github.com/re-cinq/aether/pkg/config"
 	"github.com/re-cinq/aether/pkg/log"
+	"github.com/re-cinq/aether/pkg/plugin"
 )
 
 const readHeaderTimeout = 2 * time.Second
@@ -23,10 +24,20 @@ type API struct {
 
 	addr        string
 	metricsPath string
+
+	plugins *plugin.ExportPluginSystem
+}
+
+type option func(*API)
+
+func WithExportPluginSystem(e *plugin.ExportPluginSystem) option {
+	return func(a *API) {
+		a.plugins = e
+	}
 }
 
 // New returns an instance of a configured API
-func New() *API {
+func New(opts ...option) *API {
 	api := &API{
 		metricsPath: config.AppConfig().APIConfig.MetricsPath,
 		addr: fmt.Sprintf("%s:%s",
@@ -35,6 +46,10 @@ func New() *API {
 		),
 	}
 
+	for _, o := range opts {
+		o(api)
+	}
+
 	api.setup()
 
 	return api
@@ -46,7 +61,7 @@ func (a *API) router() *mux.Router {
 	r.StrictSlash(true)
 
 	// HealthCheck
-	r.HandleFunc("/healthz", healthProbe).Methods("GET")
+	r.HandleFunc("/healthz", a.healthProbe).Methods("GET")
 
 	// Prometheus exporter
 	prometheus.MustRegister(version.NewCollector("cloud_carbon_exporter"))

pkg/api/health.go

@@ -1,13 +1,34 @@
 package api
 
 import (
+	"encoding/json"
 	"fmt"
 	"net/http"
 )
 
 // Return a 200 http status
-func healthProbe(w http.ResponseWriter, req *http.Request) {
+func (a *API) healthProbe(w http.ResponseWriter, req *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
+
+	plugins := "up"
+	var failedPlugins []string
+
+	// we need to ping every registered plugin
+	for _, p := range a.plugins.Plugins {
+		err := p.Client.Ping()
+		if err != nil {
+			plugins = "down"
+			failedPlugins = append(failedPlugins, p.Name)
+		}
+	}
+
+	failedPluginsJSON, err := json.Marshal(failedPlugins)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprintf(w, `{"error": %q}`, err)
+		return
+	}
+
 	w.WriteHeader(http.StatusOK)
-	fmt.Fprintf(w, `{"status": "up"}`)
+	fmt.Fprintf(w, `{"status": "up", "plugins": %q, failedPlugins: %s}`, plugins, failedPluginsJSON)
 }

pkg/plugin/exporter.go

@@ -91,7 +91,14 @@ func (p *ExporterPlugin) GRPCClient(
 type ExportPluginSystem struct {
 	Dir string
 
-	Plugins []Exporter
+	Plugins []*RegisteredPlugin
+}
+
+// RegisteredPlugin is all the information needed to manage a registered plugin
+type RegisteredPlugin struct {
+	Name     string
+	Exporter Exporter
+	Client   plugin.ClientProtocol
 }
 
 // Load goes to a directory and runs every binary in that directory as a plugin
@@ -136,7 +143,17 @@ func (e *ExportPluginSystem) Load(ctx context.Context) error {
 
 		p := raw.(Exporter)
 
-		e.Plugins = append(e.Plugins, p)
+		// make sure plugin is reachable
+		err = rpcClient.Ping()
+		if err != nil {
+			logger.Info("failed loading plugin", "plugin", file.Name(), "error", err)
+		}
+
+		e.Plugins = append(e.Plugins, &RegisteredPlugin{
+			Name:     file.Name(),
+			Exporter: p,
+			Client:   rpcClient,
+		})
 		logger.Info("loaded plugin", "plugin", file.Name())
 	}
 

pkg/plugin/subscriber.go

@@ -36,7 +36,7 @@ func (p *PluginHandler) SendToExporters(ctx context.Context, e *bus.Event) {
 	}
 
 	for i := range p.exporters.Plugins {
-		err := p.exporters.Plugins[i].Send(&instance)
+		err := p.exporters.Plugins[i].Exporter.Send(&instance)
 		if err != nil {
 			logger.Error("exporting instance to plugin failed", "error", err)
 		}