Add collector health status

changelog/unreleased/pr-26840.toml

@@ -0,0 +1,5 @@
+type = "a"
+message = "Expose Collector health in the user interface."
+
+issues = ["26800"]
+pulls = ["26840"]

graylog2-server/src/main/java/org/graylog/collectors/CollectorInstanceService.java

@@ -35,15 +35,22 @@
 import com.mongodb.client.model.Updates;
 import com.mongodb.client.result.InsertOneResult;
 import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import org.apache.commons.lang3.StringUtils;
+import org.bson.BsonDocument;
+import org.bson.BsonDocumentWriter;
 import org.bson.BsonType;
 import org.bson.Document;
+import org.bson.codecs.Codec;
+import org.bson.codecs.EncoderContext;
 import org.bson.conversions.Bson;
 import org.graylog.collectors.db.Attribute;
+import org.graylog.collectors.db.CollectorHealthDTO;
 import org.graylog.collectors.db.CollectorInstanceDTO;
 import org.graylog.collectors.db.CollectorInstanceReport;
+import org.graylog.collectors.db.ComponentHealthDTO;
 import org.graylog.collectors.events.CollectorInstanceCertsChangedEvent;
 import org.graylog.collectors.opamp.IssuedCertificate;
 import org.graylog2.database.MongoCollection;
@@ -85,6 +92,7 @@
 import static org.graylog.collectors.db.CollectorInstanceDTO.FIELD_CERTIFICATES_ROTATED_AT;
 import static org.graylog.collectors.db.CollectorInstanceDTO.FIELD_ENROLLMENT_TOKEN_ID;
 import static org.graylog.collectors.db.CollectorInstanceDTO.FIELD_FLEET_ID;
+import static org.graylog.collectors.db.CollectorInstanceDTO.FIELD_HEALTH;
 import static org.graylog.collectors.db.CollectorInstanceDTO.FIELD_IDENTIFYING_ATTRIBUTES;
 import static org.graylog.collectors.db.CollectorInstanceDTO.FIELD_INSTANCE_UID;
 import static org.graylog.collectors.db.CollectorInstanceDTO.FIELD_ISSUING_CA_ID;
@@ -114,6 +122,7 @@ public class CollectorInstanceService {
     private final MongoCollection<CollectorInstanceDTO> collection;
     private final MongoPaginationHelper<CollectorInstanceDTO> paginationHelper;
     private final com.mongodb.client.MongoCollection<MinimalCollectorInstanceDTO> projectedCollection;
+    private final Codec<ReportUpdateValues> reportUpdateValuesCodec;
 
     private final MongoCollections mongoCollections;
     private final ClusterEventBus clusterEventBus;
@@ -123,6 +132,7 @@ public class CollectorInstanceService {
     public CollectorInstanceService(MongoCollections mongoCollections, ClusterEventBus clusterEventBus, Clock clock) {
         collection = mongoCollections.collection(COLLECTION_NAME, CollectorInstanceDTO.class);
         projectedCollection = mongoCollections.nonEntityCollection(COLLECTION_NAME, MinimalCollectorInstanceDTO.class);
+        reportUpdateValuesCodec = mongoCollections.getCodecFor(ReportUpdateValues.class);
         paginationHelper = mongoCollections.paginationHelper(collection);
         this.mongoCollections = mongoCollections;
         this.clusterEventBus = clusterEventBus;
@@ -174,33 +184,42 @@ public CollectorInstanceService(MongoCollections mongoCollections, ClusterEventB
      * @throws IllegalArgumentException when the instance is not enrolled
      */
     public MinimalCollectorInstanceDTO updateFromReport(CollectorInstanceReport update) {
-        final List<Bson> updateOps = new ArrayList<>();
-
-        updateOps.add(set(FIELD_LAST_SEEN, Date.from(update.lastSeen())));
-        updateOps.add(set(FIELD_MESSAGE_SEQ_NUM, update.messageSeqNum()));
-        updateOps.add(set(FIELD_CAPABILITIES, update.capabilities()));
+        final var encodedValues = encodeStructuredReportValues(update);
+        // We are using an aggregation pipeline for the update so we can atomically update the health status.
+        // The aggregation pipeline allows us to use conditions to compare values in the existing document.
+        // (see #addHealthUpdate method)
+        // Dates and numbers are BSON-native and only need protection from expression evaluation.
+        final var setFields = new Document()
+                .append(FIELD_LAST_SEEN, literal(Date.from(update.lastSeen())))
+                .append(FIELD_MESSAGE_SEQ_NUM, literal(update.messageSeqNum()))
+                .append(FIELD_CAPABILITIES, literal(update.capabilities()));
         if (update.lastProcessedTxnSeq().isPresent()) {
-            updateOps.add(set(FIELD_LAST_PROCESSED_TXN_SEQ, update.lastProcessedTxnSeq().getAsLong()));
+            setFields.append(FIELD_LAST_PROCESSED_TXN_SEQ, literal(update.lastProcessedTxnSeq().getAsLong()));
         }
         if (update.identifyingAttributes().isPresent()) {
-            updateOps.add(set(FIELD_IDENTIFYING_ATTRIBUTES, update.identifyingAttributes().get()));
+            setFields.append(FIELD_IDENTIFYING_ATTRIBUTES, literal(encodedValues.get(FIELD_IDENTIFYING_ATTRIBUTES)));
         }
         if (update.nonIdentifyingAttributes().isPresent()) {
-            updateOps.add(set(FIELD_NON_IDENTIFYING_ATTRIBUTES, update.nonIdentifyingAttributes().get()));
+            setFields.append(FIELD_NON_IDENTIFYING_ATTRIBUTES, literal(encodedValues.get(FIELD_NON_IDENTIFYING_ATTRIBUTES)));
         }
+        addHealthUpdate(setFields, update, encodedValues);
+
+        final var projectedFields = List.of(
+                Projections.include(FIELD_MESSAGE_SEQ_NUM, FIELD_LAST_PROCESSED_TXN_SEQ, FIELD_FLEET_ID),
+                Projections.elemMatch(FIELD_NON_IDENTIFYING_ATTRIBUTES, Filters.eq(Attribute.FIELD_KEY, OS_TYPE_KEY))
+        );
 
         // we request the ReturnDocument.BEFORE here to avoid having to load the previous document in full just
         // to retrieve the previous `message_seq_num`, which we need to determine what to do next.
         // the result is not the full CollectorInstanceDTO as we have it, but the minimal set of fields necessary to
         // determine next steps
-        final var previousInstanceDto = projectedCollection.findOneAndUpdate(Filters.eq(FIELD_INSTANCE_UID, update.instanceUid()),
-                combine(updateOps),
+        final var previousInstanceDto = projectedCollection.findOneAndUpdate(
+                Filters.eq(FIELD_INSTANCE_UID, update.instanceUid()),
+                List.of(new Document("$set", setFields)),
                 new FindOneAndUpdateOptions()
                         .returnDocument(ReturnDocument.BEFORE)
-                        .projection(Projections.fields(
-                                Projections.include(FIELD_MESSAGE_SEQ_NUM, FIELD_LAST_PROCESSED_TXN_SEQ, FIELD_FLEET_ID),
-                                Projections.elemMatch(FIELD_NON_IDENTIFYING_ATTRIBUTES, Filters.eq(Attribute.FIELD_KEY, OS_TYPE_KEY))
-                        )));
+                        .projection(Projections.fields(projectedFields))
+        );
 
         if (previousInstanceDto == null) {
             // If there was no existing document, the instance was not enrolled.
@@ -210,6 +229,56 @@ public MinimalCollectorInstanceDTO updateFromReport(CollectorInstanceReport upda
         return previousInstanceDto;
     }
 
+    private void addHealthUpdate(Document setFields,
+                                 CollectorInstanceReport update,
+                                 BsonDocument encodedValues) {
+        // If the Collector stops reporting health information, we remove the health field so we don't show
+        // outdated information.
+        if (!update.reportsHealth()) {
+            setFields.append(FIELD_HEALTH, "$$REMOVE");
+            return;
+        }
+
+        // Preserve the stored snapshot when the capability is present but this report omits health; otherwise replace
+        // it and calculate the transition timestamp from the pre-update value in the same atomic operation.
+        update.health().ifPresent(health -> {
+            final var componentHealthField = f("%s.%s", FIELD_HEALTH, CollectorHealthDTO.FIELD_COMPONENT_HEALTH);
+            final var healthyChangedAtField = f("%s.%s", FIELD_HEALTH, CollectorHealthDTO.FIELD_HEALTHY_CHANGED_AT);
+            final var now = Date.from(clock.instant());
+            final var healthyUnchanged = new Document("$eq", List.of(
+                    "$" + componentHealthField + "." + ComponentHealthDTO.HEALTHY_FIELD,
+                    health.healthy()
+            ));
+            final var existingTimestampOrNow = new Document("$ifNull", List.of(
+                    "$" + healthyChangedAtField,
+                    now
+            ));
+
+            setFields.append(componentHealthField, literal(encodedValues.get(FIELD_HEALTH)));
+            setFields.append(healthyChangedAtField, new Document("$cond", List.of(healthyUnchanged, existingTimestampOrNow, now)));
+        });
+    }
+
+    private BsonDocument encodeStructuredReportValues(CollectorInstanceReport update) {
+        // The aggregation pipeline is built as a generic Document. Its codec handles BSON-native values, but it
+        // does not know how Graylog's Attribute and ComponentHealthDTO types must be stored. Encode those values
+        // through a MongoJack-backed wrapper first so custom field names and serializers are applied. The caller
+        // then extracts only the present fields and inserts their already-encoded BsonValues as pipeline literals.
+        final var values = new ReportUpdateValues(
+                update.identifyingAttributes().orElse(null),
+                update.nonIdentifyingAttributes().orElse(null),
+                update.health().orElse(null)
+        );
+        try (final var writer = new BsonDocumentWriter(new BsonDocument())) {
+            reportUpdateValuesCodec.encode(writer, values, EncoderContext.builder().build());
+            return writer.getDocument();
+        }
+    }
+
+    private static Document literal(Object value) {
+        return new Document("$literal", value);
+    }
+
     /**
      * Updates an existing collector instance to a new fleet id.
      *
@@ -656,6 +725,11 @@ private static CollectorOSType extractOSType(Stream<Attribute> attributes) {
                 .orElse(CollectorOSType.UNKNOWN);
     }
 
+    record ReportUpdateValues(
+            @JsonProperty(FIELD_IDENTIFYING_ATTRIBUTES) @Nullable List<Attribute> identifyingAttributes,
+            @JsonProperty(FIELD_NON_IDENTIFYING_ATTRIBUTES) @Nullable List<Attribute> nonIdentifyingAttributes,
+            @JsonProperty(FIELD_HEALTH) @Nullable ComponentHealthDTO health) {}
+
     public record MinimalCollectorInstanceDTO(@Id @JsonProperty(FIELD_ID) String id,
                                               @JsonProperty(FIELD_FLEET_ID) String fleetId,
                                               @JsonProperty(FIELD_MESSAGE_SEQ_NUM) long messageSeqNum,

graylog2-server/src/main/java/org/graylog/collectors/db/CollectorHealthDTO.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.collectors.db;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.google.auto.value.AutoValue;
+import org.graylog2.jackson.MongoInstantDeserializer;
+import org.graylog2.jackson.MongoInstantSerializer;
+
+import java.time.Instant;
+
+@AutoValue
+@JsonDeserialize(builder = CollectorHealthDTO.Builder.class)
+public abstract class CollectorHealthDTO {
+
+    public static final String FIELD_COMPONENT_HEALTH = "component_health";
+    public static final String FIELD_HEALTHY_CHANGED_AT = "healthy_changed_at";
+
+    @JsonProperty(FIELD_HEALTHY_CHANGED_AT)
+    @JsonSerialize(using = MongoInstantSerializer.class)
+    public abstract Instant healthyChangedAt();
+
+    @JsonProperty(FIELD_COMPONENT_HEALTH)
+    public abstract ComponentHealthDTO componentHealth();
+
+    public static Builder builder() {
+        return Builder.create();
+    }
+
+    @AutoValue.Builder
+    public abstract static class Builder {
+
+        @JsonCreator
+        public static Builder create() {
+            return new AutoValue_CollectorHealthDTO.Builder();
+        }
+
+        @JsonProperty(FIELD_HEALTHY_CHANGED_AT)
+        @JsonDeserialize(using = MongoInstantDeserializer.class)
+        public abstract Builder healthyChangedAt(Instant healthyChangedAt);
+
+        @JsonProperty(FIELD_COMPONENT_HEALTH)
+        public abstract Builder componentHealth(ComponentHealthDTO componentHealth);
+
+        public abstract CollectorHealthDTO build();
+    }
+}

graylog2-server/src/main/java/org/graylog/collectors/db/CollectorInstanceDTO.java

@@ -59,6 +59,7 @@ public abstract class CollectorInstanceDTO implements BuildableMongoEntity<Colle
     public static final String FIELD_NON_IDENTIFYING_ATTRIBUTES = "non_identifying_attributes";
     public static final String FIELD_LAST_PROCESSED_TXN_SEQ = "last_processed_txn_seq";
     public static final String FIELD_ENROLLMENT_TOKEN_ID = "enrollment_token_id";
+    public static final String FIELD_HEALTH = "health";
 
 
     @JsonProperty(FIELD_INSTANCE_UID)
@@ -136,6 +137,9 @@ public abstract class CollectorInstanceDTO implements BuildableMongoEntity<Colle
     @JsonProperty(FIELD_ENROLLMENT_TOKEN_ID)
     public abstract String enrollmentTokenId();
 
+    @JsonProperty(FIELD_HEALTH)
+    public abstract Optional<CollectorHealthDTO> health();
+
     public static Builder builder() {
         return AutoValue_CollectorInstanceDTO.Builder.create();
     }
@@ -222,6 +226,9 @@ public static Builder create() {
         @JsonProperty(FIELD_ENROLLMENT_TOKEN_ID)
         public abstract Builder enrollmentTokenId(String enrollmentTokenId);
 
+        @JsonProperty(FIELD_HEALTH)
+        public abstract Builder health(@Nullable CollectorHealthDTO health);
+
         public abstract CollectorInstanceDTO build();
     }
 }

graylog2-server/src/main/java/org/graylog/collectors/db/CollectorInstanceReport.java

@@ -31,7 +31,9 @@ public record CollectorInstanceReport(
         OptionalLong lastProcessedTxnSeq,
         Instant lastSeen,
         Optional<List<Attribute>> identifyingAttributes,
-        Optional<List<Attribute>> nonIdentifyingAttributes
+        Optional<List<Attribute>> nonIdentifyingAttributes,
+        boolean reportsHealth,
+        Optional<ComponentHealthDTO> health
 ) {
     @AutoBuilder
     public interface Builder {
@@ -49,11 +51,17 @@ public interface Builder {
 
         Builder nonIdentifyingAttributes(List<Attribute> nonIdentifyingAttributes);
 
+        Builder reportsHealth(boolean reportsHealth);
+
+        Builder health(ComponentHealthDTO health);
+
         CollectorInstanceReport build();
 
     }
 
     public static Builder builder() {
-        return new AutoBuilder_CollectorInstanceReport_Builder().lastSeen(Instant.now());
+        return new AutoBuilder_CollectorInstanceReport_Builder()
+                .lastSeen(Instant.now())
+                .reportsHealth(false);
     }
 }

graylog2-server/src/main/java/org/graylog/collectors/db/ComponentHealthDTO.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.collectors.db;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.google.auto.value.AutoValue;
+import jakarta.annotation.Nullable;
+import org.graylog2.jackson.MongoInstantDeserializer;
+import org.graylog2.jackson.MongoInstantSerializer;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+
+@AutoValue
+@JsonDeserialize(builder = ComponentHealthDTO.Builder.class)
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+public abstract class ComponentHealthDTO {
+    public static final String HEALTHY_FIELD = "healthy";
+
+    @JsonProperty(HEALTHY_FIELD)
+    public abstract boolean healthy();
+
+    @JsonProperty("start_time")
+    @JsonSerialize(contentUsing = MongoInstantSerializer.class)
+    public abstract Optional<Instant> startTime();
+
+    @JsonProperty("last_error")
+    public abstract Optional<String> lastError();
+
+    @JsonProperty("status")
+    public abstract Optional<String> status();
+
+    @JsonProperty("status_time")
+    @JsonSerialize(contentUsing = MongoInstantSerializer.class)
+    public abstract Optional<Instant> statusTime();
+
+    @JsonProperty("components")
+    public abstract Map<String, ComponentHealthDTO> components();
+
+    public static Builder builder() {
+        return Builder.create();
+    }
+
+    @AutoValue.Builder
+    public abstract static class Builder {
+
+        @JsonCreator
+        public static Builder create() {
+            return new AutoValue_ComponentHealthDTO.Builder()
+                    .components(Collections.emptyMap());
+        }
+
+        @JsonProperty(HEALTHY_FIELD)
+        public abstract Builder healthy(boolean healthy);
+
+        @JsonProperty("start_time")
+        @JsonDeserialize(using = MongoInstantDeserializer.class)
+        public abstract Builder startTime(@Nullable Instant startTime);
+
+        @JsonProperty("last_error")
+        public abstract Builder lastError(@Nullable String lastError);
+
+        @JsonProperty("status")
+        public abstract Builder status(@Nullable String status);
+
+        @JsonProperty("status_time")
+        @JsonDeserialize(using = MongoInstantDeserializer.class)
+        public abstract Builder statusTime(@Nullable Instant statusTime);
+
+        @JsonProperty("components")
+        public abstract Builder components(Map<String, ComponentHealthDTO> components);
+
+        public abstract ComponentHealthDTO build();
+    }
+}