report output health

internal/pkg/dl/constants.go

@@ -16,6 +16,7 @@ const (
 	FleetPolicies          = ".fleet-policies"
 	FleetPoliciesLeader    = ".fleet-policies-leader"
 	FleetServers           = ".fleet-servers"
+	FleetOutputHealth      = "logs-fleet_server.output_health-default"
 )
 
 // Query fields

internal/pkg/dl/output_health.go

@@ -0,0 +1,54 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package dl
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/elastic/fleet-server/v7/internal/pkg/bulk"
+	"github.com/gofrs/uuid"
+)
+
+type DataStream struct {
+	Dataset   string `json:"dataset,omitempty"`
+	Type      string `json:"type,omitempty"`
+	Namespace string `json:"namespace,omitempty"`
+}
+
+type OutputHealth struct {
+	Output     string     `json:"output,omitempty"`
+	State      string     `json:"state,omitempty"`
+	Message    string     `json:"message,omitempty"`
+	Timestamp  string     `json:"@timestamp,omitempty"`
+	DataStream DataStream `json:"data_stream,omitempty"`
+}
+
+func CreateOutputHealth(ctx context.Context, bulker bulk.Bulk, doc OutputHealth) error {
+	return createOutputHealth(ctx, bulker, FleetOutputHealth, doc)
+}
+
+func createOutputHealth(ctx context.Context, bulker bulk.Bulk, index string, doc OutputHealth) error {
+	if doc.Timestamp == "" {
+		doc.Timestamp = time.Now().UTC().Format(time.RFC3339)
+	}
+	doc.DataStream = DataStream{
+		Dataset:   "fleet_server.output_health",
+		Type:      "logs",
+		Namespace: "default",
+	}
+	body, err := json.Marshal(doc)
+	if err != nil {
+		return err
+	}
+
+	id, err := uuid.NewV4()
+	if err != nil {
+		return err
+	}
+	_, err = bulker.Create(ctx, index, id.String(), body, bulk.WithRefresh())
+	return err
+}

internal/pkg/policy/policy_output.go

@@ -15,6 +15,7 @@ import (
 	"github.com/rs/zerolog"
 	"go.elastic.co/apm/v2"
 
+	"github.com/elastic/elastic-agent-client/v7/pkg/client"
 	"github.com/elastic/fleet-server/v7/internal/pkg/apikey"
 	"github.com/elastic/fleet-server/v7/internal/pkg/bulk"
 	"github.com/elastic/fleet-server/v7/internal/pkg/dl"
@@ -260,17 +261,36 @@ func (p *Output) prepareElasticsearch(
 		ctx := zlog.WithContext(ctx)
 		outputAPIKey, err :=
 			generateOutputAPIKey(ctx, outputBulker, agent.Id, p.Name, p.Role.Raw)
-			// reporting output error status to self monitor and not returning the error to keep fleet-server running
+
+		// reporting output health and not returning the error to keep fleet-server running
 		if outputAPIKey == nil && p.Type == OutputTypeRemoteElasticsearch {
 			if err != nil {
-				zerolog.Ctx(ctx).Warn().Err(err).Msg("Could not create API key in remote ES")
+				doc := dl.OutputHealth{
+					Output:  p.Name,
+					State:   client.UnitStateDegraded.String(),
+					Message: fmt.Sprintf("remote ES could not create API key due to error: %s", err.Error()),
+				}
+				zerolog.Ctx(ctx).Warn().Err(err).Msg(doc.Message)
+
+				if err := dl.CreateOutputHealth(ctx, bulker, doc); err != nil {
+					zlog.Error().Err(err).Msg("error writing output health")
+				}
 			}
 
 			// replace type remote_elasticsearch with elasticsearch as agent doesn't recognize remote_elasticsearch
 			outputMap[p.Name][FieldOutputType] = OutputTypeElasticsearch
 			// remove the service token from the agent policy sent to the agent
 			delete(outputMap[p.Name], FieldOutputServiceToken)
 			return nil
+		} else if p.Type == OutputTypeRemoteElasticsearch {
+			doc := dl.OutputHealth{
+				Output:  p.Name,
+				State:   client.UnitStateHealthy.String(),
+				Message: "",
+			}
+			if err := dl.CreateOutputHealth(ctx, bulker, doc); err != nil {
+				zlog.Error().Err(err).Msg("create output health")
+			}
 		}
 		if err != nil {
 			return fmt.Errorf("failed generate output API key: %w", err)
@@ -304,7 +324,7 @@ func (p *Output) prepareElasticsearch(
 		// Using painless script to append the old keys to the history
 		body, err := renderUpdatePainlessScript(p.Name, fields)
 		if err != nil {
-			return fmt.Errorf("could no tupdate painless script: %w", err)
+			return fmt.Errorf("could not update painless script: %w", err)
 		}
 
 		if err = bulker.Update(ctx, dl.FleetAgents, agent.Id, body, bulk.WithRefresh(), bulk.WithRetryOnConflict(3)); err != nil {

internal/pkg/policy/self.go

@@ -218,6 +218,8 @@ func (m *selfMonitorT) updateState(ctx context.Context) (client.UnitState, error
 		return client.UnitStateStarting, nil
 	}
 
+	reportOutputHealth(ctx, m.bulker, m.log)
+
 	state := client.UnitStateHealthy
 	extendMsg := ""
 	var payload map[string]interface{}
@@ -253,6 +255,32 @@ func (m *selfMonitorT) updateState(ctx context.Context) (client.UnitState, error
 	return state, nil
 }
 
+func reportOutputHealth(ctx context.Context, bulker bulk.Bulk, logger zerolog.Logger) {
+	//pinging logic
+	bulkerMap := bulker.GetBulkerMap()
+	for outputName, outputBulker := range bulkerMap {
+		doc := dl.OutputHealth{
+			Output:  outputName,
+			State:   client.UnitStateHealthy.String(),
+			Message: "",
+		}
+		res, err := outputBulker.Client().Ping(outputBulker.Client().Ping.WithContext(ctx))
+		if err != nil {
+			doc.State = client.UnitStateDegraded.String()
+			doc.Message = fmt.Sprintf("remote ES is not reachable due to error: %s", err.Error())
+			logger.Error().Err(err).Msg(doc.Message)
+
+		} else if res.StatusCode != 200 {
+			doc.State = client.UnitStateDegraded.String()
+			doc.Message = fmt.Sprintf("remote ES is not reachable due to unexpected status code %d", res.StatusCode)
+			logger.Error().Err(err).Msg(doc.Message)
+		}
+		if err := dl.CreateOutputHealth(ctx, bulker, doc); err != nil {
+			logger.Error().Err(err).Msg("error writing output health")
+		}
+	}
+}
+
 func HasFleetServerInput(inputs []map[string]interface{}) bool {
 	for _, input := range inputs {
 		attr, ok := input["type"].(string)