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,18 @@
 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.BsonType;
 import org.bson.Document;
 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 +88,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;
@@ -113,7 +117,7 @@ public class CollectorInstanceService {
 
     private final MongoCollection<CollectorInstanceDTO> collection;
     private final MongoPaginationHelper<CollectorInstanceDTO> paginationHelper;
-    private final com.mongodb.client.MongoCollection<MinimalCollectorInstanceDTO> projectedCollection;
+    private final com.mongodb.client.MongoCollection<ReportUpdateState> reportUpdateCollection;
 
     private final MongoCollections mongoCollections;
     private final ClusterEventBus clusterEventBus;
@@ -122,7 +126,7 @@ public class CollectorInstanceService {
     @Inject
     public CollectorInstanceService(MongoCollections mongoCollections, ClusterEventBus clusterEventBus, Clock clock) {
         collection = mongoCollections.collection(COLLECTION_NAME, CollectorInstanceDTO.class);
-        projectedCollection = mongoCollections.nonEntityCollection(COLLECTION_NAME, MinimalCollectorInstanceDTO.class);
+        reportUpdateCollection = mongoCollections.nonEntityCollection(COLLECTION_NAME, ReportUpdateState.class);
         paginationHelper = mongoCollections.paginationHelper(collection);
         this.mongoCollections = mongoCollections;
         this.clusterEventBus = clusterEventBus;
@@ -189,25 +193,61 @@ public MinimalCollectorInstanceDTO updateFromReport(CollectorInstanceReport upda
             updateOps.add(set(FIELD_NON_IDENTIFYING_ATTRIBUTES, update.nonIdentifyingAttributes().get()));
         }
 
+        final var projectedFields = new ArrayList<Bson>();
+        projectedFields.add(Projections.include(FIELD_MESSAGE_SEQ_NUM, FIELD_LAST_PROCESSED_TXN_SEQ, FIELD_FLEET_ID));
+        projectedFields.add(
+                Projections.elemMatch(FIELD_NON_IDENTIFYING_ATTRIBUTES, Filters.eq(Attribute.FIELD_KEY, OS_TYPE_KEY)));
+        // on an incoming health report, request the original health field, to determine if it needs to be updated
+        if (update.health().isPresent()) {
+            projectedFields.add(Projections.include(FIELD_HEALTH));
+        }
+
         // 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()),
+        final var previousInstanceDto = reportUpdateCollection.findOneAndUpdate(Filters.eq(FIELD_INSTANCE_UID, update.instanceUid()),
                 combine(updateOps),
                 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.
             throw new IllegalArgumentException("Instance not enrolled: " + update.instanceUid());
         }
 
-        return previousInstanceDto;
+        if (update.health().isPresent()) {
+            final var previousComponentHealth = Optional.ofNullable(previousInstanceDto.health())
+                    .map(CollectorHealthDTO::componentHealth);
+
+            // If health changed, update the document and set the changed_at timestamp correctly
+            if (!update.health().equals(previousComponentHealth)) {
+                final var healthUpdates = new ArrayList<Bson>();
+                healthUpdates.add(set(f("%s.%s", FIELD_HEALTH, CollectorHealthDTO.FIELD_COMPONENT_HEALTH),
+                        update.health().get()));
+
+                final var healthyChanged = !update.health().map(ComponentHealthDTO::healthy)
+                        .equals(previousComponentHealth.map(ComponentHealthDTO::healthy));
+                if (healthyChanged) {
+                    healthUpdates.add(
+                            set(f("%s.%s", FIELD_HEALTH, CollectorHealthDTO.FIELD_HEALTHY_CHANGED_AT),
+                                    Date.from(clock.instant())));
+                }
+
+                collection.updateOne(Filters.eq(FIELD_INSTANCE_UID, update.instanceUid()),
+                        Updates.combine(healthUpdates));
+            }
+        }
+
+        return new MinimalCollectorInstanceDTO(
+                previousInstanceDto.id(),
+                previousInstanceDto.fleetId(),
+                previousInstanceDto.messageSeqNum(),
+                previousInstanceDto.lastProcessTxnSeq(),
+                previousInstanceDto.nonIdentifyingAttributes()
+        );
     }
 
     /**
@@ -656,6 +696,14 @@ private static CollectorOSType extractOSType(Stream<Attribute> attributes) {
                 .orElse(CollectorOSType.UNKNOWN);
     }
 
+    record ReportUpdateState(@Id @JsonProperty(FIELD_ID) String id,
+                             @JsonProperty(FIELD_FLEET_ID) String fleetId,
+                             @JsonProperty(FIELD_MESSAGE_SEQ_NUM) long messageSeqNum,
+                             @JsonProperty(FIELD_LAST_PROCESSED_TXN_SEQ) long lastProcessTxnSeq,
+                             @JsonProperty(FIELD_NON_IDENTIFYING_ATTRIBUTES) List<Attribute> nonIdentifyingAttributes,
+                             // health will only be populated by our projection if the incoming report included health
+                             @JsonProperty(FIELD_HEALTH) @Nullable CollectorHealthDTO 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,8 @@ public record CollectorInstanceReport(
         OptionalLong lastProcessedTxnSeq,
         Instant lastSeen,
         Optional<List<Attribute>> identifyingAttributes,
-        Optional<List<Attribute>> nonIdentifyingAttributes
+        Optional<List<Attribute>> nonIdentifyingAttributes,
+        Optional<ComponentHealthDTO> health
 ) {
     @AutoBuilder
     public interface Builder {
@@ -49,6 +50,8 @@ public interface Builder {
 
         Builder nonIdentifyingAttributes(List<Attribute> nonIdentifyingAttributes);
 
+        Builder health(ComponentHealthDTO health);
+
         CollectorInstanceReport build();
 
     }

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

@@ -0,0 +1,94 @@
+/*
+ * 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 {
+
+    @JsonProperty("healthy")
+    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")
+        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();
+    }
+}