diff --git a/graylog2-server/src/main/java/org/graylog2/bindings/PeriodicalBindings.java b/graylog2-server/src/main/java/org/graylog2/bindings/PeriodicalBindings.java index e470ce92ef18..fad39f0fba80 100644 --- a/graylog2-server/src/main/java/org/graylog2/bindings/PeriodicalBindings.java +++ b/graylog2-server/src/main/java/org/graylog2/bindings/PeriodicalBindings.java @@ -36,6 +36,7 @@ import org.graylog2.periodical.LeaderPresenceCheckPeriodical; import org.graylog2.periodical.NodeMetricPeriodical; import org.graylog2.periodical.NodePingThread; +import org.graylog2.periodical.StaleInputRuntimeStateCleanup; import org.graylog2.periodical.OrphanedTokenCleaner; import org.graylog2.periodical.SearchVersionCheckPeriodical; import org.graylog2.periodical.ThrottleStateUpdaterThread; @@ -74,5 +75,6 @@ protected void configure() { periodicalBinder.addBinding().to(ExpiredTokenCleaner.class); periodicalBinder.addBinding().to(OrphanedTokenCleaner.class); periodicalBinder.addBinding().to(NodeMetricPeriodical.class); + periodicalBinder.addBinding().to(StaleInputRuntimeStateCleanup.class); } } diff --git a/graylog2-server/src/main/java/org/graylog2/inputs/InputRuntimeStatusProvider.java b/graylog2-server/src/main/java/org/graylog2/inputs/InputRuntimeStatusProvider.java index 659e7305ba8a..6d7238fda23c 100644 --- a/graylog2-server/src/main/java/org/graylog2/inputs/InputRuntimeStatusProvider.java +++ b/graylog2-server/src/main/java/org/graylog2/inputs/InputRuntimeStatusProvider.java @@ -17,52 +17,28 @@ package org.graylog2.inputs; import jakarta.inject.Inject; -import jakarta.inject.Named; import jakarta.inject.Singleton; -import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodeService; import org.graylog2.database.filtering.ComputedFieldProvider; +import org.graylog2.inputs.persistence.InputStateService; import org.graylog2.plugin.IOState; -import org.graylog2.plugin.system.NodeId; -import org.graylog2.rest.RemoteInterfaceProvider; -import org.graylog2.rest.resources.system.inputs.RemoteInputStatesResource; -import org.graylog2.shared.inputs.InputRegistry; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import retrofit2.Response; -import java.util.HashMap; import java.util.HashSet; import java.util.Locale; import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; -import java.util.stream.Collectors; /** - * Provides filtering support for input runtime status by querying InputRegistry across all cluster nodes. + * Provides filtering support for input runtime status by querying MongoDB. *

- * This provider enables filtering inputs by their actual runtime state (RUNNING, FAILED, STOPPED, etc.) - * rather than just the desired_state stored in the database. It aggregates runtime status information - * from all nodes in the cluster to provide a cluster-wide view. - *

- *

- * Optimizations: - *

- * + * Runtime state is persisted to MongoDB by {@link org.graylog2.inputs.InputStateListener}, + * enabling efficient cluster-wide queries by state without HTTP fan-out. */ @Singleton public class InputRuntimeStatusProvider implements ComputedFieldProvider { private static final Logger LOG = LoggerFactory.getLogger(InputRuntimeStatusProvider.class); private static final String FIELD_NAME = "runtime_status"; - private static final long CACHE_TTL_MS = 5_000; public static final Map> STATUS_GROUPS = Map.of( "RUNNING", Set.of(IOState.Type.RUNNING), @@ -78,29 +54,14 @@ public class InputRuntimeStatusProvider implements ComputedFieldProvider { "FAILED", "Failed" ); - private final NodeService nodeService; - private final RemoteInterfaceProvider remoteInterfaceProvider; - private final InputRegistry inputRegistry; + private final InputStateService runtimeStateService; private final InputService inputService; - private final NodeId nodeId; - private final ExecutorService executorService; - - private volatile Map> cachedStatuses; - private volatile long cacheTimestamp = 0; @Inject - public InputRuntimeStatusProvider(NodeService nodeService, - RemoteInterfaceProvider remoteInterfaceProvider, - InputRegistry inputRegistry, - InputService inputService, - NodeId nodeId, - @Named("proxiedRequestsExecutorService") ExecutorService executorService) { - this.nodeService = nodeService; - this.remoteInterfaceProvider = remoteInterfaceProvider; - this.inputRegistry = inputRegistry; + public InputRuntimeStatusProvider(InputStateService runtimeStateService, + InputService inputService) { + this.runtimeStateService = runtimeStateService; this.inputService = inputService; - this.nodeId = nodeId; - this.executorService = executorService; } @Override @@ -112,27 +73,16 @@ public Set getMatchingIds(String filterValue, String authToken) { return Set.of(); } - // For NOT_RUNNING: query desired_state from DB since stopped inputs are absent from the InputRegistry + // For NOT_RUNNING: query desired_state from DB since stopped inputs have no runtime state if ("NOT_RUNNING".equals(key)) { final Set matching = inputService.findIdsByDesiredState(IOState.Type.STOPPED); LOG.debug("Found {} inputs with runtime_status group={}", matching.size(), key); return matching; } - - final Set targetStrings = targetStates.stream() - .map(IOState.Type::toString) - .collect(Collectors.toSet()); - - final Map> allStatuses = getClusterStatuses(authToken); final Set matching = new HashSet<>(); - for (Map.Entry> entry : allStatuses.entrySet()) { - for (String status : entry.getValue()) { - if (targetStrings.contains(status)) { - matching.add(entry.getKey()); - break; - } - } + for (IOState.Type type : targetStates) { + matching.addAll(runtimeStateService.getByState(type)); } LOG.debug("Found {} inputs with runtime_status group={}", matching.size(), key); @@ -143,63 +93,4 @@ public Set getMatchingIds(String filterValue, String authToken) { public String getFieldName() { return FIELD_NAME; } - - private Map> getClusterStatuses(String authToken) { - long now = System.currentTimeMillis(); - if (cachedStatuses != null && (now - cacheTimestamp) < CACHE_TTL_MS) { - return cachedStatuses; - } - synchronized (this) { - // Double-check after acquiring lock - if (cachedStatuses != null && (System.currentTimeMillis() - cacheTimestamp) < CACHE_TTL_MS) { - return cachedStatuses; - } - cachedStatuses = fetchAllClusterStatuses(authToken); - cacheTimestamp = System.currentTimeMillis(); - return cachedStatuses; - } - } - - private Map> fetchAllClusterStatuses(String authToken) { - final Map> result = new ConcurrentHashMap<>(); - final String localNodeId = nodeId.getNodeId(); - - // 1. Local node: read InputRegistry directly (no HTTP) - inputRegistry.getStatusesByInputId().forEach((inputId, status) -> - result.computeIfAbsent(inputId, k -> ConcurrentHashMap.newKeySet()).add(status)); - - // 2. Remote nodes: parallel HTTP calls - final Map activeNodes = nodeService.allActive(); - final Map>> futures = new HashMap<>(); - - for (Map.Entry entry : activeNodes.entrySet()) { - if (entry.getKey().equals(localNodeId)) { - continue; // skip local node, already handled above - } - final Node node = entry.getValue(); - futures.put(entry.getKey(), executorService.submit(() -> { - final RemoteInputStatesResource remote = remoteInterfaceProvider.get( - node, authToken, RemoteInputStatesResource.class); - final Response> response = remote.getLocalStatuses().execute(); - if (response.isSuccessful() && response.body() != null) { - return response.body(); - } - LOG.debug("Failed to get response from node {}: {}", node.getNodeId(), response.code()); - return Map.of(); - })); - } - - // 3. Collect results with timeout - for (Map.Entry>> entry : futures.entrySet()) { - try { - final Map nodeStatuses = entry.getValue().get(30, TimeUnit.SECONDS); - nodeStatuses.forEach((id, status) -> - result.computeIfAbsent(id, k -> ConcurrentHashMap.newKeySet()).add(status)); - } catch (Exception e) { - LOG.debug("Error fetching input states from node {}: {}", entry.getKey(), e.getMessage()); - } - } - - return result; - } } diff --git a/graylog2-server/src/main/java/org/graylog2/inputs/InputStateListener.java b/graylog2-server/src/main/java/org/graylog2/inputs/InputStateListener.java index 6541e951c40c..1cc5cb0d67e2 100644 --- a/graylog2-server/src/main/java/org/graylog2/inputs/InputStateListener.java +++ b/graylog2-server/src/main/java/org/graylog2/inputs/InputStateListener.java @@ -19,6 +19,7 @@ import com.google.common.eventbus.EventBus; import com.google.common.eventbus.Subscribe; import org.apache.commons.lang3.ObjectUtils; +import org.graylog2.inputs.persistence.InputStateService; import org.graylog2.notifications.Notification; import org.graylog2.notifications.NotificationService; import org.graylog2.plugin.IOState; @@ -43,15 +44,18 @@ public class InputStateListener { private final NotificationService notificationService; private final ActivityWriter activityWriter; private final ServerStatus serverStatus; + private final InputStateService inputStateService; @Inject public InputStateListener(EventBus eventBus, NotificationService notificationService, ActivityWriter activityWriter, - ServerStatus serverStatus) { + ServerStatus serverStatus, + InputStateService inputStateService) { this.notificationService = notificationService; this.activityWriter = activityWriter; this.serverStatus = serverStatus; + this.inputStateService = inputStateService; eventBus.register(this); } @@ -88,5 +92,24 @@ public void inputStateChanged(IOStateChangedEvent event) { LOG.debug("Input State of {} changed: {} -> {}", input.toIdentifier(), event.oldState(), event.newState()); LOG.info("Input {} is now {}", input.toIdentifier(), event.newState()); + + final String inputId = input.getId(); + if (inputId != null) { + try { + if (event.newState() == IOState.Type.TERMINATED) { + inputStateService.removeState(inputId); + } else { + inputStateService.upsertState( + inputId, + state.getState(), + state.getStartedAt(), + state.getLastFailedAt(), + state.getDetailedMessage()); + } + } catch (Exception e) { + LOG.warn("Failed to persist runtime state for input {}: {}", inputId, e.getMessage()); + LOG.debug("Exception details:", e); + } + } } } diff --git a/graylog2-server/src/main/java/org/graylog2/inputs/persistence/InputStateDto.java b/graylog2-server/src/main/java/org/graylog2/inputs/persistence/InputStateDto.java new file mode 100644 index 000000000000..55c186058e49 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/inputs/persistence/InputStateDto.java @@ -0,0 +1,113 @@ +/* + * 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 + * . + */ +package org.graylog2.inputs.persistence; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.google.auto.value.AutoValue; +import org.graylog2.database.MongoEntity; +import org.joda.time.DateTime; +import org.mongojack.Id; +import org.mongojack.ObjectId; + +import javax.annotation.Nullable; + +@AutoValue +@JsonDeserialize(builder = InputStateDto.Builder.class) +public abstract class InputStateDto implements MongoEntity { + static final String FIELD_ID = "id"; + static final String FIELD_INPUT_ID = "input_id"; + static final String FIELD_NODE_ID = "node_id"; + static final String FIELD_STATE = "state"; + static final String FIELD_STARTED_AT = "started_at"; + static final String FIELD_LAST_FAILED_AT = "last_failed_at"; + static final String FIELD_DETAILED_MESSAGE = "detailed_message"; + static final String FIELD_UPDATED_AT = "updated_at"; + + @Id + @ObjectId + @Nullable + @JsonProperty(FIELD_ID) + public abstract String id(); + + @JsonProperty(FIELD_INPUT_ID) + public abstract String inputId(); + + @JsonProperty(FIELD_NODE_ID) + public abstract String nodeId(); + + @JsonProperty(FIELD_STATE) + public abstract String state(); + + @Nullable + @JsonProperty(FIELD_STARTED_AT) + public abstract DateTime startedAt(); + + @Nullable + @JsonProperty(FIELD_LAST_FAILED_AT) + public abstract DateTime lastFailedAt(); + + @Nullable + @JsonProperty(FIELD_DETAILED_MESSAGE) + public abstract String detailedMessage(); + + @JsonProperty(FIELD_UPDATED_AT) + public abstract DateTime updatedAt(); + + public static Builder builder() { + return Builder.create(); + } + + public abstract Builder toBuilder(); + + @AutoValue.Builder + public static abstract class Builder { + @JsonCreator + public static Builder create() { + return new AutoValue_InputStateDto.Builder(); + } + + @Id + @ObjectId + @JsonProperty(FIELD_ID) + public abstract Builder id(String id); + + @JsonProperty(FIELD_INPUT_ID) + public abstract Builder inputId(String inputId); + + @JsonProperty(FIELD_NODE_ID) + public abstract Builder nodeId(String nodeId); + + @JsonProperty(FIELD_STATE) + public abstract Builder state(String state); + + @JsonProperty(FIELD_STARTED_AT) + public abstract Builder startedAt(DateTime startedAt); + + @JsonProperty(FIELD_LAST_FAILED_AT) + public abstract Builder lastFailedAt(DateTime lastFailedAt); + + @JsonProperty(FIELD_DETAILED_MESSAGE) + public abstract Builder detailedMessage(String detailedMessage); + + @JsonProperty(FIELD_UPDATED_AT) + public abstract Builder updatedAt(DateTime updatedAt); + + public abstract InputStateDto build(); + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/inputs/persistence/InputStateService.java b/graylog2-server/src/main/java/org/graylog2/inputs/persistence/InputStateService.java new file mode 100644 index 000000000000..b71cc4454e3e --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/inputs/persistence/InputStateService.java @@ -0,0 +1,152 @@ +/* + * 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 + * . + */ +package org.graylog2.inputs.persistence; + +import com.mongodb.client.model.Filters; +import com.mongodb.client.model.FindOneAndUpdateOptions; +import com.mongodb.client.model.IndexOptions; +import com.mongodb.client.model.Indexes; +import com.mongodb.client.model.ReturnDocument; +import com.mongodb.client.model.Updates; +import jakarta.inject.Inject; +import jakarta.inject.Singleton; +import org.bson.conversions.Bson; +import org.graylog2.database.MongoCollection; +import org.graylog2.database.MongoCollections; +import org.graylog2.database.utils.MongoUtils; +import org.graylog2.plugin.IOState; +import org.graylog2.plugin.Tools; +import org.graylog2.plugin.system.NodeId; +import org.joda.time.DateTime; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.graylog2.inputs.persistence.InputStateDto.FIELD_DETAILED_MESSAGE; +import static org.graylog2.inputs.persistence.InputStateDto.FIELD_INPUT_ID; +import static org.graylog2.inputs.persistence.InputStateDto.FIELD_LAST_FAILED_AT; +import static org.graylog2.inputs.persistence.InputStateDto.FIELD_NODE_ID; +import static org.graylog2.inputs.persistence.InputStateDto.FIELD_STARTED_AT; +import static org.graylog2.inputs.persistence.InputStateDto.FIELD_STATE; +import static org.graylog2.inputs.persistence.InputStateDto.FIELD_UPDATED_AT; + +@Singleton +public class InputStateService { + private static final Logger LOG = LoggerFactory.getLogger(InputStateService.class); + private static final String COLLECTION_NAME = "input_runtime_states"; + + private final MongoCollection collection; + private final String thisNodeId; + + @Inject + public InputStateService(MongoCollections mongoCollections, NodeId nodeId) { + this.collection = mongoCollections.collection(COLLECTION_NAME, InputStateDto.class); + this.thisNodeId = nodeId.getNodeId(); + + collection.createIndex( + Indexes.ascending(FIELD_INPUT_ID, FIELD_NODE_ID), + new IndexOptions().unique(true)); + collection.createIndex(Indexes.ascending(FIELD_NODE_ID)); + collection.createIndex(Indexes.ascending(FIELD_STATE)); + } + + public void upsertState(String inputId, IOState.Type state, + @Nullable DateTime startedAt, + @Nullable DateTime lastFailedAt, + @Nullable String detailedMessage) { + final Bson filter = Filters.and( + Filters.eq(FIELD_INPUT_ID, inputId), + Filters.eq(FIELD_NODE_ID, thisNodeId)); + + final var updates = new java.util.ArrayList(); + updates.add(Updates.set(FIELD_INPUT_ID, inputId)); + updates.add(Updates.set(FIELD_NODE_ID, thisNodeId)); + updates.add(Updates.set(FIELD_STATE, state.toString())); + updates.add(Updates.set(FIELD_UPDATED_AT, Tools.nowUTC())); + + if (startedAt != null) { + updates.add(Updates.set(FIELD_STARTED_AT, startedAt)); + } + if (lastFailedAt != null) { + updates.add(Updates.set(FIELD_LAST_FAILED_AT, lastFailedAt)); + } + if (detailedMessage != null) { + updates.add(Updates.set(FIELD_DETAILED_MESSAGE, detailedMessage)); + } else { + updates.add(Updates.unset(FIELD_DETAILED_MESSAGE)); + } + + collection.findOneAndUpdate(filter, Updates.combine(updates), + new FindOneAndUpdateOptions().upsert(true).returnDocument(ReturnDocument.AFTER)); + } + + public void removeState(String inputId) { + collection.deleteOne(Filters.and( + Filters.eq(FIELD_INPUT_ID, inputId), + Filters.eq(FIELD_NODE_ID, thisNodeId))); + } + + public void removeAllForNode() { + final long deleted = collection.deleteMany(Filters.eq(FIELD_NODE_ID, thisNodeId)).getDeletedCount(); + LOG.debug("Removed {} runtime state documents for node {}", deleted, thisNodeId); + } + + public void removeAllForNode(String nodeId) { + final long deleted = collection.deleteMany(Filters.eq(FIELD_NODE_ID, nodeId)).getDeletedCount(); + if (deleted > 0) { + LOG.debug("Removed {} stale runtime state documents for node {}", deleted, nodeId); + } + } + + + public Map> getClusterStatuses() { + try (var stream = MongoUtils.stream(collection.find())) { + return stream.collect(Collectors.groupingBy( + InputStateDto::inputId, + HashMap::new, + Collectors.mapping(InputStateDto::state, Collectors.toSet()) + )); + } + } + + public Set getByState(IOState.Type state) { + try (var stream = MongoUtils.stream(collection.find(Filters.eq(FIELD_STATE, state.toString())))) { + return stream.map(InputStateDto::inputId).collect(Collectors.toSet()); + } + } + + public Set getByStates(Collection states) { + List stateStrings = states.stream().map(IOState.Type::toString).toList(); + Bson filter = Filters.in(FIELD_STATE, stateStrings); + try (var stream = MongoUtils.stream(collection.find(filter))) { + return stream.collect(Collectors.toSet()); + } + } + + public Set getDistinctNodeIds() { + try (var stream = MongoUtils.stream(collection.distinct(FIELD_NODE_ID, String.class))) { + return stream.collect(Collectors.toSet()); + } + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/periodical/StaleInputRuntimeStateCleanup.java b/graylog2-server/src/main/java/org/graylog2/periodical/StaleInputRuntimeStateCleanup.java new file mode 100644 index 000000000000..d93ab422bd2d --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/periodical/StaleInputRuntimeStateCleanup.java @@ -0,0 +1,102 @@ +/* + * 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 + * . + */ +package org.graylog2.periodical; + +import jakarta.inject.Inject; +import org.graylog2.cluster.NodeService; +import org.graylog2.inputs.persistence.InputStateService; +import org.graylog2.plugin.periodical.Periodical; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Set; + +/** + * Periodically removes runtime state documents for nodes that are no longer active in the cluster. + * Runs on the leader node only. + */ +public class StaleInputRuntimeStateCleanup extends Periodical { + private static final Logger LOG = LoggerFactory.getLogger(StaleInputRuntimeStateCleanup.class); + + private final InputStateService runtimeStateService; + private final NodeService nodeService; + + @Inject + public StaleInputRuntimeStateCleanup(InputStateService runtimeStateService, + NodeService nodeService) { + this.runtimeStateService = runtimeStateService; + this.nodeService = nodeService; + } + + @Override + public void doRun() { + try { + final Set activeNodeIds = nodeService.allActive().keySet(); + final Set stateNodeIds = runtimeStateService.getDistinctNodeIds(); + + for (String nodeId : stateNodeIds) { + if (!activeNodeIds.contains(nodeId)) { + LOG.debug("Cleaning up stale runtime state documents for inactive node {}", nodeId); + runtimeStateService.removeAllForNode(nodeId); + } + } + } catch (Exception e) { + LOG.warn("Error during stale input runtime state cleanup: {}", e.getMessage()); + LOG.debug("Exception details:", e); + } + } + + @Override + public boolean runsForever() { + return false; + } + + @Override + public boolean stopOnGracefulShutdown() { + return true; + } + + @Override + public boolean leaderOnly() { + return true; + } + + @Override + public boolean startOnThisNode() { + return true; + } + + @Override + public boolean isDaemon() { + return true; + } + + @Override + public int getInitialDelaySeconds() { + return 60; + } + + @Override + public int getPeriodSeconds() { + return 60; + } + + @Override + protected Logger getLogger() { + return LOG; + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/rest/resources/system/inputs/InputStatesResource.java b/graylog2-server/src/main/java/org/graylog2/rest/resources/system/inputs/InputStatesResource.java index 4e0ea148d11e..0976fd146e35 100644 --- a/graylog2-server/src/main/java/org/graylog2/rest/resources/system/inputs/InputStatesResource.java +++ b/graylog2-server/src/main/java/org/graylog2/rest/resources/system/inputs/InputStatesResource.java @@ -34,10 +34,12 @@ import jakarta.ws.rs.Produces; import jakarta.ws.rs.core.MediaType; import org.apache.shiro.authz.annotation.RequiresAuthentication; +import org.apache.shiro.authz.annotation.RequiresPermissions; import org.graylog2.audit.AuditEventTypes; import org.graylog2.audit.jersey.AuditEvent; import org.graylog2.inputs.Input; import org.graylog2.inputs.InputService; +import org.graylog2.inputs.persistence.InputStateService; import org.graylog2.plugin.IOState; import org.graylog2.plugin.database.ValidationException; import org.graylog2.plugin.inputs.MessageInput; @@ -53,6 +55,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Map; import java.util.Set; import java.util.stream.Collectors; @@ -65,16 +68,19 @@ public class InputStatesResource extends AbstractInputsResource { private final InputRegistry inputRegistry; private final EventBus serverEventBus; private final InputService inputService; + private final InputStateService inputStateService; @Inject public InputStatesResource(InputRegistry inputRegistry, EventBus serverEventBus, InputService inputService, - MessageInputFactory messageInputFactory) { + MessageInputFactory messageInputFactory, + InputStateService inputStateService) { super(messageInputFactory.getAvailableInputs()); this.inputRegistry = inputRegistry; this.serverEventBus = serverEventBus; this.inputService = inputService; + this.inputStateService = inputStateService; } @GET @@ -90,11 +96,12 @@ public InputStatesList list() { } @GET - @Path("/local") + @Path("/summary") @Timed - @Operation(summary = "Get runtime status of all inputs on this node (for cluster-wide filtering)") - public java.util.Map getLocalStatuses() { - return this.inputRegistry.getStatusesByInputId(); + @Operation(summary = "Get lightweight cluster input state summary from DB") + @RequiresPermissions(RestPermissions.INPUTS_READ) + public Map> summary() { + return inputStateService.getClusterStatuses(); } @GET diff --git a/graylog2-server/src/main/java/org/graylog2/rest/resources/system/inputs/RemoteInputStatesResource.java b/graylog2-server/src/main/java/org/graylog2/rest/resources/system/inputs/RemoteInputStatesResource.java index ab81f131bebb..da7b90ed80fa 100644 --- a/graylog2-server/src/main/java/org/graylog2/rest/resources/system/inputs/RemoteInputStatesResource.java +++ b/graylog2-server/src/main/java/org/graylog2/rest/resources/system/inputs/RemoteInputStatesResource.java @@ -30,9 +30,6 @@ public interface RemoteInputStatesResource { @GET("system/inputstates") Call list(); - @GET("system/inputstates/local") - Call> getLocalStatuses(); - @PUT("system/inputstates/{inputId}") Call start(@Path("inputId") String inputId); diff --git a/graylog2-server/src/main/java/org/graylog2/shared/initializers/InputSetupService.java b/graylog2-server/src/main/java/org/graylog2/shared/initializers/InputSetupService.java index df2e1fb16997..7b0cc9e3378e 100644 --- a/graylog2-server/src/main/java/org/graylog2/shared/initializers/InputSetupService.java +++ b/graylog2-server/src/main/java/org/graylog2/shared/initializers/InputSetupService.java @@ -21,6 +21,9 @@ import com.google.common.eventbus.Subscribe; import com.google.common.util.concurrent.AbstractExecutionThreadService; import com.google.common.util.concurrent.Uninterruptibles; +import jakarta.inject.Inject; +import jakarta.inject.Singleton; +import org.graylog2.inputs.persistence.InputStateService; import org.graylog2.plugin.IOState; import org.graylog2.plugin.inputs.MessageInput; import org.graylog2.plugin.lifecycles.Lifecycle; @@ -29,9 +32,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -42,16 +42,19 @@ public class InputSetupService extends AbstractExecutionThreadService { private final InputRegistry inputRegistry; private final EventBus eventBus; private final InputLauncher inputLauncher; + private final InputStateService runtimeStateService; private final CountDownLatch startLatch = new CountDownLatch(1); private final CountDownLatch stopLatch = new CountDownLatch(1); private AtomicReference previousLifecycle = new AtomicReference<>(Lifecycle.UNINITIALIZED); @Inject - public InputSetupService(InputRegistry inputRegistry, EventBus eventBus, InputLauncher inputLauncher) { + public InputSetupService(InputRegistry inputRegistry, EventBus eventBus, InputLauncher inputLauncher, + InputStateService runtimeStateService) { this.inputRegistry = inputRegistry; this.eventBus = eventBus; this.inputLauncher = inputLauncher; + this.runtimeStateService = runtimeStateService; } @Override @@ -122,6 +125,13 @@ protected void shutDown() throws Exception { s.stop(); } } + + try { + runtimeStateService.removeAllForNode(); + } catch (Exception e) { + LOG.warn("Failed to clean up input state documents during shutdown: {}", e.getMessage()); + } + LOG.debug("Stopped InputSetupService"); } } diff --git a/graylog2-server/src/main/java/org/graylog2/shared/inputs/InputLauncher.java b/graylog2-server/src/main/java/org/graylog2/shared/inputs/InputLauncher.java index 86a29ab6bccd..27444db4285a 100644 --- a/graylog2-server/src/main/java/org/graylog2/shared/inputs/InputLauncher.java +++ b/graylog2-server/src/main/java/org/graylog2/shared/inputs/InputLauncher.java @@ -84,12 +84,11 @@ public IOState launch(final MessageInput input) { final IOState inputState; if (inputRegistry.getInputState(input.getId()) == null) { + inputState = inputStateFactory.create(input); + inputRegistry.add(inputState); if (featureFlags.isOn("SETUP_MODE") && input.getDesiredState() == IOState.Type.SETUP) { - inputState = inputStateFactory.create(input, IOState.Type.SETUP); - } else { - inputState = inputStateFactory.create(input); + inputState.setState(IOState.Type.SETUP); } - inputRegistry.add(inputState); } else { inputState = inputRegistry.getInputState(input.getId()); switch (inputState.getState()) { diff --git a/graylog2-server/src/main/java/org/graylog2/shared/inputs/InputRegistry.java b/graylog2-server/src/main/java/org/graylog2/shared/inputs/InputRegistry.java index f0a0ca7a108f..fa3361d6d603 100644 --- a/graylog2-server/src/main/java/org/graylog2/shared/inputs/InputRegistry.java +++ b/graylog2-server/src/main/java/org/graylog2/shared/inputs/InputRegistry.java @@ -18,68 +18,40 @@ import com.google.common.collect.ImmutableSet; -import com.google.common.eventbus.EventBus; -import com.google.common.eventbus.Subscribe; import jakarta.inject.Inject; import jakarta.inject.Singleton; import org.graylog2.plugin.IOState; -import org.graylog2.plugin.events.inputs.IOStateChangedEvent; import org.graylog2.plugin.inputs.MessageInput; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.Collection; -import java.util.HashMap; -import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.stream.Stream; @Singleton public class InputRegistry { private static final Logger LOG = LoggerFactory.getLogger(InputRegistry.class); - private final InputStateCache cache = new InputStateCache(); + private final ConcurrentMap> inputStates = new ConcurrentHashMap<>(); @Inject - public InputRegistry(EventBus eventBus) { - eventBus.register(this); - } - - @Subscribe - public void onIOStateChanged(IOStateChangedEvent event) { - final IOState changedState = event.changedState(); - final String inputId = changedState.getStoppable().getId(); - - if (!cache.contains(inputId)) { - return; - } - - cache.updateState(inputId, event.oldState(), event.newState()); - } - - public Map getStatusesByInputId() { - return cache.getStatusesByInputId(); - } - - public Set getInputIdsByState(IOState.Type state) { - return cache.getIdsByState(state); + public InputRegistry() { } public Set> getInputStates() { - return ImmutableSet.copyOf(cache.getAll()); + return ImmutableSet.copyOf(inputStates.values()); } public IOState getInputState(String inputId) { - return cache.get(inputId); + return inputStates.get(inputId); } public Set> getRunningInputs() { - Set runningIds = cache.getIdsByState(IOState.Type.RUNNING); ImmutableSet.Builder> builder = ImmutableSet.builder(); - for (String id : runningIds) { - IOState state = cache.get(id); - if (state != null) { + for (IOState state : inputStates.values()) { + if (state.getState() == IOState.Type.RUNNING) { builder.add(state); } } @@ -87,7 +59,7 @@ public Set> getRunningInputs() { } public boolean hasTypeRunning(Class klazz) { - for (IOState inputState : cache.getAll()) { + for (IOState inputState : inputStates.values()) { if (inputState.getStoppable().getClass().equals(klazz)) { return true; } @@ -96,7 +68,13 @@ public boolean hasTypeRunning(Class klazz) { } public int runningCount() { - return cache.getIdsByState(IOState.Type.RUNNING).size(); + int count = 0; + for (IOState state : inputStates.values()) { + if (state.getState() == IOState.Type.RUNNING) { + count++; + } + } + return count; } public boolean remove(MessageInput input) { @@ -104,7 +82,7 @@ public boolean remove(MessageInput input) { input.terminate(); if (inputState != null) { inputState.setState(IOState.Type.TERMINATED); - cache.remove(input.getId()); + inputStates.remove(input.getId()); } return inputState != null; } @@ -115,7 +93,7 @@ public boolean remove(IOState inputState) { } public IOState stop(MessageInput input) { - IOState inputState = cache.get(input.getId()); + IOState inputState = inputStates.get(input.getId()); if (inputState != null) { inputState.setState(IOState.Type.STOPPING); @@ -137,73 +115,12 @@ public void setup(IOState inputState) { } public boolean add(IOState messageInputIOState) { - return cache.add(messageInputIOState); + final String inputId = messageInputIOState.getStoppable().getId(); + return inputStates.putIfAbsent(inputId, messageInputIOState) == null; } public Stream> stream() { - return cache.getAll().stream(); + return inputStates.values().stream(); } - class InputStateCache { - private final ConcurrentHashMap> byId = new ConcurrentHashMap<>(); - private final ConcurrentHashMap> byState = new ConcurrentHashMap<>(); - - private IOState get(String inputId) { - return byId.get(inputId); - } - - boolean contains(String inputId) { - return byId.containsKey(inputId); - } - - private Set getIdsByState(IOState.Type state) { - final Set ids = byState.get(state); - return ids != null ? Set.copyOf(ids) : Set.of(); - } - - private Collection> getAll() { - return byId.values(); - } - - private boolean add(IOState ioState) { - final String inputId = ioState.getStoppable().getId(); - final IOState previous = byId.putIfAbsent(inputId, ioState); - if (previous != null) { - return false; - } - byState.computeIfAbsent(ioState.getState(), k -> ConcurrentHashMap.newKeySet()) - .add(inputId); - return true; - } - - private IOState remove(String inputId) { - final IOState removed = byId.remove(inputId); - if (removed != null) { - for (Set ids : byState.values()) { - ids.remove(inputId); - } - } - return removed; - } - - private void updateState(String inputId, IOState.Type oldState, IOState.Type newState) { - final Set oldSet = byState.get(oldState); - if (oldSet != null) { - oldSet.remove(inputId); - } - byState.computeIfAbsent(newState, k -> ConcurrentHashMap.newKeySet()) - .add(inputId); - } - - private Map getStatusesByInputId() { - final Map result = new HashMap<>(); - for (Map.Entry> entry : byState.entrySet()) { - final String stateStr = entry.getKey().toString(); - for (String inputId : entry.getValue()) { - result.put(inputId, stateStr); - } - } - return result; - } - } } diff --git a/graylog2-server/src/test/java/org/graylog2/contentpacks/facades/InputFacadeTest.java b/graylog2-server/src/test/java/org/graylog2/contentpacks/facades/InputFacadeTest.java index 619995affd28..43cc0eadcebc 100644 --- a/graylog2-server/src/test/java/org/graylog2/contentpacks/facades/InputFacadeTest.java +++ b/graylog2-server/src/test/java/org/graylog2/contentpacks/facades/InputFacadeTest.java @@ -143,7 +143,7 @@ public void setUp(MongoCollections mongoCollections) throws Exception { final ExtractorFactory extractorFactory = new ExtractorFactory(metricRegistry, grokPatternRegistry, lookupTableService); final ConverterFactory converterFactory = new ConverterFactory(lookupTableService); inputService = new InputServiceImpl(mongoCollections, extractorFactory, converterFactory, messageInputFactory, clusterEventBus, new ObjectMapperProvider().get()); - final InputRegistry inputRegistry = new InputRegistry(clusterBus); + final InputRegistry inputRegistry = new InputRegistry(); Set pluginMetaData = new HashSet<>(); Map> inputFactories = new HashMap<>(); final FakeHttpMessageInput.Factory fakeHttpMessageInputFactory = mock(FakeHttpMessageInput.Factory.class); diff --git a/graylog2-server/src/test/java/org/graylog2/inputs/InputRuntimeStatusProviderTest.java b/graylog2-server/src/test/java/org/graylog2/inputs/InputRuntimeStatusProviderTest.java index fdf105a8955d..43c723bb1b8b 100644 --- a/graylog2-server/src/test/java/org/graylog2/inputs/InputRuntimeStatusProviderTest.java +++ b/graylog2-server/src/test/java/org/graylog2/inputs/InputRuntimeStatusProviderTest.java @@ -16,79 +16,39 @@ */ package org.graylog2.inputs; -import com.google.common.util.concurrent.ThreadFactoryBuilder; -import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodeService; +import org.graylog2.inputs.persistence.InputStateService; import org.graylog2.plugin.IOState; -import org.graylog2.plugin.system.NodeId; -import org.graylog2.rest.RemoteInterfaceProvider; -import org.graylog2.rest.resources.system.inputs.RemoteInputStatesResource; -import org.graylog2.shared.inputs.InputRegistry; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; -import retrofit2.Call; -import retrofit2.Response; -import java.io.IOException; -import java.util.Map; import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.lenient; -import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) class InputRuntimeStatusProviderTest { - private static final String LOCAL_NODE_ID = "local-node-1"; private static final String AUTH_TOKEN = "test-token"; private InputRuntimeStatusProvider provider; @Mock - private NodeService nodeService; - - @Mock - private RemoteInterfaceProvider remoteInterfaceProvider; - - @Mock - private InputRegistry inputRegistry; + private InputStateService runtimeStateService; @Mock private InputService inputService; - @Mock - private NodeId nodeId; - - private ExecutorService executorService; - @BeforeEach void setUp() { - executorService = Executors.newFixedThreadPool(4, new ThreadFactoryBuilder() - .setNameFormat("input-runtime-status-test-%d") - .build()); - lenient().when(nodeId.getNodeId()).thenReturn(LOCAL_NODE_ID); - - provider = new InputRuntimeStatusProvider( - nodeService, - remoteInterfaceProvider, - inputRegistry, - inputService, - nodeId, - executorService - ); + provider = new InputRuntimeStatusProvider(runtimeStateService, inputService); } @Test @@ -103,127 +63,22 @@ void returnsEmptySetForInvalidStatus() { } @Test - void localRegistryUsedDirectly() { - when(inputRegistry.getStatusesByInputId()).thenReturn(Map.of( - "input-1", "RUNNING", - "input-2", "FAILED" - )); - when(nodeService.allActive()).thenReturn(Map.of(LOCAL_NODE_ID, mock(Node.class))); - - // Should find the running input without any remote calls - final Set running = provider.getMatchingIds("RUNNING", AUTH_TOKEN); - assertEquals(Set.of("input-1"), running); - - // Verify no remote HTTP calls were made (local node is skipped) - verify(remoteInterfaceProvider, never()).get(any(Node.class), any(), eq(RemoteInputStatesResource.class)); - } - - @Test - void remoteNodesQueriedInParallel() throws Exception { - when(inputRegistry.getStatusesByInputId()).thenReturn(Map.of()); - - // Set up two remote nodes - Node remoteNode1 = mock(Node.class); - Node remoteNode2 = mock(Node.class); - - when(nodeService.allActive()).thenReturn(Map.of( - "remote-1", remoteNode1, - "remote-2", remoteNode2 - )); - - // Mock remote calls - mockRemoteCall(remoteNode1, Map.of("input-a", "RUNNING", "input-b", "FAILED")); - mockRemoteCall(remoteNode2, Map.of("input-c", "RUNNING")); - - final Set running = provider.getMatchingIds("RUNNING", AUTH_TOKEN); - assertEquals(Set.of("input-a", "input-c"), running); - } - - @Test - void globalInputMatchesAnyNodeStatus() throws Exception { - // Same input on local node is RUNNING, on remote node is FAILED - when(inputRegistry.getStatusesByInputId()) - .thenReturn(Map.of("global-input", "RUNNING")); - - Node remoteNode = mock(Node.class); - when(nodeService.allActive()).thenReturn(Map.of( - LOCAL_NODE_ID, mock(Node.class), - "remote-1", remoteNode - )); - - mockRemoteCall(remoteNode, Map.of("global-input", "FAILED")); - - // Create a new provider so cache is fresh - provider = new InputRuntimeStatusProvider( - nodeService, remoteInterfaceProvider, inputRegistry, inputService, nodeId, executorService); - - // Should match RUNNING (from local) — cache has both RUNNING and FAILED for global-input - final Set running = provider.getMatchingIds("RUNNING", AUTH_TOKEN); - assertTrue(running.contains("global-input")); - - // Should also match FAILED (from remote) using same cache - final Set failed = provider.getMatchingIds("FAILED", AUTH_TOKEN); - assertTrue(failed.contains("global-input")); - } - - @Test - void cacheReuseAvoidsDuplicateFetch() { - when(inputRegistry.getStatusesByInputId()).thenReturn(Map.of("input-1", "RUNNING")); - when(nodeService.allActive()).thenReturn(Map.of(LOCAL_NODE_ID, mock(Node.class))); - - // First call populates cache - provider.getMatchingIds("RUNNING", AUTH_TOKEN); - // Second call should reuse cache - provider.getMatchingIds("RUNNING", AUTH_TOKEN); - - // getStatusesByInputId() should only be called once (cached on second call) - verify(inputRegistry, times(1)).getStatusesByInputId(); - } - - @Test - void nodeFailureHandledGracefully() throws Exception { - when(inputRegistry.getStatusesByInputId()).thenReturn(Map.of()); - - Node failingNode = mock(Node.class); - Node goodNode = mock(Node.class); - - when(nodeService.allActive()).thenReturn(Map.of( - "failing-node", failingNode, - "good-node", goodNode - )); - - // Failing node throws IOException - RemoteInputStatesResource failingRemote = mock(RemoteInputStatesResource.class); - @SuppressWarnings("unchecked") - Call> failingCall = mock(Call.class); - when(failingCall.execute()).thenThrow(new IOException("Connection refused")); - when(failingRemote.getLocalStatuses()).thenReturn(failingCall); - when(remoteInterfaceProvider.get(failingNode, AUTH_TOKEN, RemoteInputStatesResource.class)) - .thenReturn(failingRemote); - - // Good node returns data - mockRemoteCall(goodNode, Map.of("input-1", "RUNNING")); + void runningGroupQueriesMongoDb() { + when(runtimeStateService.getByState(IOState.Type.RUNNING)) + .thenReturn(Set.of("input-1", "input-3")); final Set running = provider.getMatchingIds("RUNNING", AUTH_TOKEN); - assertEquals(Set.of("input-1"), running); - } - - @Test - void handlesUpperCaseAndLowerCaseStatus() { - // Invalid after uppercase conversion should return empty - assertTrue(provider.getMatchingIds("invalid", AUTH_TOKEN).isEmpty()); - assertTrue(provider.getMatchingIds("INVALID", AUTH_TOKEN).isEmpty()); + assertEquals(Set.of("input-1", "input-3"), running); } @Test void failedGroupMatchesFailingAndInvalidConfiguration() { - when(inputRegistry.getStatusesByInputId()).thenReturn(Map.of( - "input-1", "FAILING", - "input-2", "INVALID_CONFIGURATION", - "input-3", "RUNNING", - "input-4", "FAILED" - )); - when(nodeService.allActive()).thenReturn(Map.of(LOCAL_NODE_ID, mock(Node.class))); + when(runtimeStateService.getByState(IOState.Type.FAILED)) + .thenReturn(Set.of("input-4")); + when(runtimeStateService.getByState(IOState.Type.FAILING)) + .thenReturn(Set.of("input-1")); + when(runtimeStateService.getByState(IOState.Type.INVALID_CONFIGURATION)) + .thenReturn(Set.of("input-2")); final Set failed = provider.getMatchingIds("FAILED", AUTH_TOKEN); assertEquals(Set.of("input-1", "input-2", "input-4"), failed); @@ -237,44 +92,35 @@ void notRunningGroupQueriesDbForStoppedDesiredState() { final Set notRunning = provider.getMatchingIds("NOT_RUNNING", AUTH_TOKEN); assertEquals(Set.of("input-1", "input-5"), notRunning); - // Should NOT query the registry or remote nodes for NOT_RUNNING - verify(inputRegistry, never()).getStatusesByInputId(); - verify(nodeService, never()).allActive(); - } - - @Test - void runningGroupStillUsesRegistry() { - when(inputRegistry.getStatusesByInputId()).thenReturn(Map.of("input-1", "RUNNING")); - when(nodeService.allActive()).thenReturn(Map.of(LOCAL_NODE_ID, mock(Node.class))); - - final Set running = provider.getMatchingIds("RUNNING", AUTH_TOKEN); - assertEquals(Set.of("input-1"), running); - - // Should NOT query the DB for RUNNING - verify(inputService, never()).findIdsByDesiredState(any()); + // Should NOT query runtime state service for NOT_RUNNING + verify(runtimeStateService, never()).getByState(any()); } @Test void setupGroupMatchesSetupInitializedStarting() { - when(inputRegistry.getStatusesByInputId()).thenReturn(Map.of( - "input-1", "SETUP", - "input-2", "INITIALIZED", - "input-3", "STARTING", - "input-4", "RUNNING" - )); - when(nodeService.allActive()).thenReturn(Map.of(LOCAL_NODE_ID, mock(Node.class))); + when(runtimeStateService.getByState(IOState.Type.SETUP)) + .thenReturn(Set.of("input-1")); + when(runtimeStateService.getByState(IOState.Type.INITIALIZED)) + .thenReturn(Set.of("input-2")); + when(runtimeStateService.getByState(IOState.Type.STARTING)) + .thenReturn(Set.of("input-3")); final Set setup = provider.getMatchingIds("SETUP", AUTH_TOKEN); assertEquals(Set.of("input-1", "input-2", "input-3"), setup); } - @SuppressWarnings("unchecked") - private void mockRemoteCall(Node node, Map statuses) throws IOException { - RemoteInputStatesResource remote = mock(RemoteInputStatesResource.class); - Call> call = mock(Call.class); - when(call.execute()).thenReturn(Response.success(statuses)); - when(remote.getLocalStatuses()).thenReturn(call); - when(remoteInterfaceProvider.get(node, AUTH_TOKEN, RemoteInputStatesResource.class)) - .thenReturn(remote); + @Test + void handlesUpperCaseAndLowerCaseStatus() { + assertTrue(provider.getMatchingIds("invalid", AUTH_TOKEN).isEmpty()); + assertTrue(provider.getMatchingIds("INVALID", AUTH_TOKEN).isEmpty()); + } + + @Test + void runningGroupDoesNotQueryDb() { + when(runtimeStateService.getByState(IOState.Type.RUNNING)) + .thenReturn(Set.of("input-1")); + + provider.getMatchingIds("RUNNING", AUTH_TOKEN); + verify(inputService, never()).findIdsByDesiredState(any()); } } diff --git a/graylog2-server/src/test/java/org/graylog2/inputs/persistence/InputStateServiceTest.java b/graylog2-server/src/test/java/org/graylog2/inputs/persistence/InputStateServiceTest.java new file mode 100644 index 000000000000..98bb60410046 --- /dev/null +++ b/graylog2-server/src/test/java/org/graylog2/inputs/persistence/InputStateServiceTest.java @@ -0,0 +1,203 @@ +/* + * 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 + * . + */ +package org.graylog2.inputs.persistence; + +import org.graylog.testing.mongodb.MongoDBExtension; +import org.graylog2.database.MongoCollections; +import org.graylog2.plugin.IOState; +import org.graylog2.plugin.system.SimpleNodeId; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.junit.jupiter.MockitoExtension; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; + +import java.util.Map; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; + +@ExtendWith(MockitoExtension.class) +@ExtendWith(MongoDBExtension.class) +@MockitoSettings(strictness = Strictness.WARN) +class InputStateServiceTest { + + private static final String NODE_1 = "node-1"; + private static final String NODE_2 = "node-2"; + + private InputStateService serviceNode1; + private InputStateService serviceNode2; + + @BeforeEach + void setUp(MongoCollections mongoCollections) { + serviceNode1 = new InputStateService(mongoCollections, new SimpleNodeId(NODE_1)); + serviceNode2 = new InputStateService(mongoCollections, new SimpleNodeId(NODE_2)); + } + + @Test + void upsertAndGetByState() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + serviceNode1.upsertState("input-2", IOState.Type.FAILED, now, now, "connection refused"); + + assertThat(serviceNode1.getByState(IOState.Type.RUNNING)).containsExactly("input-1"); + assertThat(serviceNode1.getByState(IOState.Type.FAILED)).containsExactly("input-2"); + assertThat(serviceNode1.getByState(IOState.Type.STOPPED)).isEmpty(); + } + + @Test + void upsertOverwritesExistingState() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("input-1", IOState.Type.STARTING, now, null, null); + assertThat(serviceNode1.getByState(IOState.Type.STARTING)).containsExactly("input-1"); + + serviceNode1.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + assertThat(serviceNode1.getByState(IOState.Type.RUNNING)).containsExactly("input-1"); + assertThat(serviceNode1.getByState(IOState.Type.STARTING)).isEmpty(); + } + + @Test + void clusterStatusesAggregatesAcrossNodes() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + serviceNode2.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + serviceNode1.upsertState("input-2", IOState.Type.FAILED, now, now, "error"); + + Map> statuses = serviceNode1.getClusterStatuses(); + + assertThat(statuses).containsKey("input-1"); + assertThat(statuses.get("input-1")).containsExactly("RUNNING"); + assertThat(statuses).containsKey("input-2"); + assertThat(statuses.get("input-2")).containsExactly("FAILED"); + } + + @Test + void globalInputDifferentStatesOnDifferentNodes() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("global-input", IOState.Type.RUNNING, now, null, null); + serviceNode2.upsertState("global-input", IOState.Type.FAILED, now, now, "error"); + + Map> statuses = serviceNode1.getClusterStatuses(); + assertThat(statuses.get("global-input")).containsExactlyInAnyOrder("RUNNING", "FAILED"); + + // getByState should find this input under both states + assertThat(serviceNode1.getByState(IOState.Type.RUNNING)).contains("global-input"); + assertThat(serviceNode1.getByState(IOState.Type.FAILED)).contains("global-input"); + } + + @Test + void removeState() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + assertThat(serviceNode1.getByState(IOState.Type.RUNNING)).containsExactly("input-1"); + + serviceNode1.removeState("input-1"); + assertThat(serviceNode1.getByState(IOState.Type.RUNNING)).isEmpty(); + } + + @Test + void removeStateOnlyAffectsOwnNode() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + serviceNode2.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + + serviceNode1.removeState("input-1"); + + // Node 2 still has the state + assertThat(serviceNode1.getByState(IOState.Type.RUNNING)).containsExactly("input-1"); + } + + @Test + void removeAllForNode() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + serviceNode1.upsertState("input-2", IOState.Type.RUNNING, now, null, null); + serviceNode2.upsertState("input-3", IOState.Type.RUNNING, now, null, null); + + serviceNode1.removeAllForNode(); + + // Only node 2's states remain + assertThat(serviceNode1.getByState(IOState.Type.RUNNING)).containsExactly("input-3"); + } + + @Test + void removeAllForSpecificNode() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + serviceNode2.upsertState("input-2", IOState.Type.RUNNING, now, null, null); + + // Clean up node-2 from node-1's service (stale cleanup scenario) + serviceNode1.removeAllForNode(NODE_2); + + assertThat(serviceNode1.getByState(IOState.Type.RUNNING)).containsExactly("input-1"); + } + + @Test + void getDistinctNodeIds() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + serviceNode2.upsertState("input-2", IOState.Type.RUNNING, now, null, null); + + assertThat(serviceNode1.getDistinctNodeIds()).containsExactlyInAnyOrder(NODE_1, NODE_2); + } + + @Test + void emptyCollectionReturnsEmptyResults() { + assertThat(serviceNode1.getByState(IOState.Type.RUNNING)).isEmpty(); + assertThat(serviceNode1.getClusterStatuses()).isEmpty(); + assertThat(serviceNode1.getDistinctNodeIds()).isEmpty(); + } + + @Test + void getByStatesReturnsFullDtos() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + serviceNode1.upsertState("input-2", IOState.Type.FAILED, now, now, "connection refused"); + serviceNode2.upsertState("input-3", IOState.Type.RUNNING, now, null, null); + serviceNode1.upsertState("input-4", IOState.Type.STOPPED, now, null, null); + + Set results = serviceNode1.getByStates(Set.of(IOState.Type.RUNNING, IOState.Type.FAILED)); + + assertThat(results).hasSize(3); + assertThat(results).extracting(InputStateDto::inputId) + .containsExactlyInAnyOrder("input-1", "input-2", "input-3"); + assertThat(results).allSatisfy(dto -> assertThat(dto.nodeId()).isIn(NODE_1, NODE_2)); + } + + @Test + void getByStatesReturnsEmptyForNoMatches() { + final DateTime now = DateTime.now(DateTimeZone.UTC); + + serviceNode1.upsertState("input-1", IOState.Type.RUNNING, now, null, null); + + Set results = serviceNode1.getByStates(Set.of(IOState.Type.FAILED, IOState.Type.STOPPED)); + assertThat(results).isEmpty(); + } +} diff --git a/graylog2-server/src/test/java/org/graylog2/shared/inputs/InputRegistryTest.java b/graylog2-server/src/test/java/org/graylog2/shared/inputs/InputRegistryTest.java index 8f5c08784a07..be29886e0227 100644 --- a/graylog2-server/src/test/java/org/graylog2/shared/inputs/InputRegistryTest.java +++ b/graylog2-server/src/test/java/org/graylog2/shared/inputs/InputRegistryTest.java @@ -16,111 +16,99 @@ */ package org.graylog2.shared.inputs; -import com.google.common.eventbus.EventBus; import org.graylog2.plugin.IOState; import org.graylog2.plugin.inputs.MessageInput; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.junit.jupiter.MockitoExtension; -import java.util.Map; import java.util.Set; import java.util.stream.IntStream; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +@ExtendWith(MockitoExtension.class) class InputRegistryTest { - private EventBus eventBus; private InputRegistry inputRegistry; @BeforeEach void setUp() { - eventBus = new EventBus(); - inputRegistry = new InputRegistry(eventBus); + inputRegistry = new InputRegistry(); } @Test void testThreadSafe() { - // test which was failing pretty reliably back in the days when the input registry was not threadsafe IntStream.range(0, 100).parallel().forEach(i -> { if (i % 2 == 0) { var ignored = inputRegistry.stream().toList(); } else { - //noinspection unchecked inputRegistry.add(mockIOState("input-" + i, IOState.Type.CREATED)); } }); } @Test - void addIndexesCurrentState() { + void addAndGetInputState() { IOState state = mockIOState("input-1", IOState.Type.RUNNING); - inputRegistry.add(state); + assertTrue(inputRegistry.add(state)); - Map statuses = inputRegistry.getStatusesByInputId(); - assertEquals("RUNNING", statuses.get("input-1")); - assertEquals(Set.of("input-1"), inputRegistry.getInputIdsByState(IOState.Type.RUNNING)); + assertNotNull(inputRegistry.getInputState("input-1")); + assertNull(inputRegistry.getInputState("nonexistent")); } @Test - void stateChangeUpdatesIndex() { - IOState state = new IOState<>(eventBus, mockMessageInput("input-1"), IOState.Type.CREATED); - inputRegistry.add(state); + void addDuplicateReturnsFalse() { + IOState state1 = mockIOState("input-1", IOState.Type.RUNNING); + IOState state2 = mockIOState("input-1", IOState.Type.FAILED); - assertEquals(Set.of("input-1"), inputRegistry.getInputIdsByState(IOState.Type.CREATED)); - assertTrue(inputRegistry.getInputIdsByState(IOState.Type.RUNNING).isEmpty()); - - state.setState(IOState.Type.RUNNING); - - assertEquals(Set.of("input-1"), inputRegistry.getInputIdsByState(IOState.Type.RUNNING)); - assertTrue(inputRegistry.getInputIdsByState(IOState.Type.CREATED).isEmpty()); + assertTrue(inputRegistry.add(state1)); + assertFalse(inputRegistry.add(state2)); } @Test - void multipleStateTransitionsTrackedCorrectly() { - IOState state = new IOState<>(eventBus, mockMessageInput("input-1"), IOState.Type.CREATED); - inputRegistry.add(state); - - state.setState(IOState.Type.STARTING); - state.setState(IOState.Type.RUNNING); - state.setState(IOState.Type.STOPPING); - state.setState(IOState.Type.STOPPED); - - Map statuses = inputRegistry.getStatusesByInputId(); - assertEquals("STOPPED", statuses.get("input-1")); - assertEquals(Set.of("input-1"), inputRegistry.getInputIdsByState(IOState.Type.STOPPED)); - assertTrue(inputRegistry.getInputIdsByState(IOState.Type.RUNNING).isEmpty()); - assertTrue(inputRegistry.getInputIdsByState(IOState.Type.CREATED).isEmpty()); + void getInputStatesReturnsAll() { + inputRegistry.add(mockIOState("input-1", IOState.Type.RUNNING)); + inputRegistry.add(mockIOState("input-2", IOState.Type.FAILED)); + inputRegistry.add(mockIOState("input-3", IOState.Type.RUNNING)); + + Set> states = inputRegistry.getInputStates(); + assertEquals(3, states.size()); } @Test - void getStatusesByInputIdReturnsAllInputs() { + void getRunningInputs() { inputRegistry.add(mockIOState("input-1", IOState.Type.RUNNING)); inputRegistry.add(mockIOState("input-2", IOState.Type.FAILED)); inputRegistry.add(mockIOState("input-3", IOState.Type.RUNNING)); - Map statuses = inputRegistry.getStatusesByInputId(); - assertEquals(3, statuses.size()); - assertEquals("RUNNING", statuses.get("input-1")); - assertEquals("FAILED", statuses.get("input-2")); - assertEquals("RUNNING", statuses.get("input-3")); + Set> running = inputRegistry.getRunningInputs(); + assertEquals(2, running.size()); } @Test - void getInputIdsByStateReturnsEmptyForUnknownState() { - assertTrue(inputRegistry.getInputIdsByState(IOState.Type.RUNNING).isEmpty()); + void runningCount() { + inputRegistry.add(mockIOState("input-1", IOState.Type.RUNNING)); + inputRegistry.add(mockIOState("input-2", IOState.Type.FAILED)); + inputRegistry.add(mockIOState("input-3", IOState.Type.RUNNING)); + + assertEquals(2, inputRegistry.runningCount()); } @Test - void stateChangeForNonRegisteredInputIsIgnored() { - IOState state = new IOState<>(eventBus, mockMessageInput("outside-input"), IOState.Type.CREATED); - state.setState(IOState.Type.RUNNING); + void streamReturnsAllStates() { + inputRegistry.add(mockIOState("input-1", IOState.Type.RUNNING)); + inputRegistry.add(mockIOState("input-2", IOState.Type.FAILED)); - assertTrue(inputRegistry.getStatusesByInputId().isEmpty()); - assertTrue(inputRegistry.getInputIdsByState(IOState.Type.RUNNING).isEmpty()); + assertEquals(2, inputRegistry.stream().count()); } @SuppressWarnings("unchecked") @@ -129,14 +117,7 @@ private IOState mockIOState(String inputId, IOState.Type state) { MessageInput messageInput = mock(MessageInput.class); when(messageInput.getId()).thenReturn(inputId); when(ioState.getStoppable()).thenReturn(messageInput); - when(ioState.getState()).thenReturn(state); + lenient().when(ioState.getState()).thenReturn(state); return ioState; } - - private MessageInput mockMessageInput(String inputId) { - MessageInput messageInput = mock(MessageInput.class); - when(messageInput.getId()).thenReturn(inputId); - when(messageInput.getPersistId()).thenReturn(inputId); - return messageInput; - } } diff --git a/graylog2-web-interface/src/components/inputs/InputsDotBadge.test.tsx b/graylog2-web-interface/src/components/inputs/InputsDotBadge.test.tsx index 77ffafe8e729..a103fad3f7d0 100644 --- a/graylog2-web-interface/src/components/inputs/InputsDotBadge.test.tsx +++ b/graylog2-web-interface/src/components/inputs/InputsDotBadge.test.tsx @@ -18,13 +18,11 @@ import * as React from 'react'; import { render, screen } from 'wrappedTestingLibrary'; import { asMock } from 'helpers/mocking'; -import type { InputState } from 'hooks/useInputsStates'; -import useInputsStates from 'hooks/useInputsStates'; -import type { InputSummary } from 'hooks/usePaginatedInputs'; +import useInputStateSummary from 'hooks/useInputStateSummary'; import InputsDotBadge from './InputsDotBadge'; -jest.mock('hooks/useInputsStates'); +jest.mock('hooks/useInputStateSummary'); const TEXT = 'Inputs'; @@ -34,10 +32,9 @@ describe('', () => { }); it('renders text while loading', () => { - asMock(useInputsStates).mockReturnValue({ - refetch: jest.fn(), + asMock(useInputStateSummary).mockReturnValue({ + hasProblematicInputs: false, isLoading: true, - data: undefined, }); render(); @@ -45,16 +42,10 @@ describe('', () => { expect(screen.getByText(TEXT)).toBeInTheDocument(); }); - it('renders plain text when there are no failed/failing/setup inputs', () => { - asMock(useInputsStates).mockReturnValue({ - refetch: jest.fn(), + it('renders plain text when there are no problematic inputs', () => { + asMock(useInputStateSummary).mockReturnValue({ + hasProblematicInputs: false, isLoading: false, - data: { - input1: { - nodeA: { state: 'RUNNING', id: '1', detailed_message: null, message_input: {} as InputSummary }, - nodeB: { state: 'STARTING', id: '2', detailed_message: 'Error', message_input: {} as InputSummary }, - }, - }, }); render(); @@ -64,39 +55,23 @@ describe('', () => { expect(textEl).not.toHaveAttribute('title'); }); - describe.each(['FAILED', 'FAILING', 'SETUP'])( - 'renders badge when an input state is %s', - (problemState: InputState) => { - it(`shows badge (dot) with tooltip for state ${problemState}`, () => { - asMock(useInputsStates).mockReturnValue({ - refetch: jest.fn(), - isLoading: false, - data: { - input1: { - nodeA: { state: 'RUNNING', id: '1', detailed_message: null, message_input: {} as InputSummary }, - nodeB: { state: problemState, id: '2', detailed_message: 'Error', message_input: {} as InputSummary }, - }, - }, - }); - - render(); - - const badge = screen.getByTitle(/Some inputs are in failed state or in setup mode\./i); - expect(badge).toBeInTheDocument(); - expect(badge).toHaveTextContent(TEXT); - }); - }, - ); - - it('shows badge when hasExternalIssues is true and no failed inputs', () => { - asMock(useInputsStates).mockReturnValue({ - refetch: jest.fn(), + it('shows badge when there are problematic inputs', () => { + asMock(useInputStateSummary).mockReturnValue({ + hasProblematicInputs: true, + isLoading: false, + }); + + render(); + + const badge = screen.getByTitle(/Some inputs are in failed state or in setup mode\./i); + expect(badge).toBeInTheDocument(); + expect(badge).toHaveTextContent(TEXT); + }); + + it('shows badge when hasExternalIssues is true and no problematic inputs', () => { + asMock(useInputStateSummary).mockReturnValue({ + hasProblematicInputs: false, isLoading: false, - data: { - input1: { - nodeA: { state: 'RUNNING', id: '1', detailed_message: null, message_input: {} as InputSummary }, - }, - }, }); render(); @@ -106,15 +81,10 @@ describe('', () => { expect(badge).toHaveTextContent(TEXT); }); - it('shows failed inputs title when both failed inputs and external issues exist', () => { - asMock(useInputsStates).mockReturnValue({ - refetch: jest.fn(), + it('shows failed inputs title when both problematic inputs and external issues exist', () => { + asMock(useInputStateSummary).mockReturnValue({ + hasProblematicInputs: true, isLoading: false, - data: { - input1: { - nodeA: { state: 'FAILED', id: '1', detailed_message: 'Error', message_input: {} as InputSummary }, - }, - }, }); render(); diff --git a/graylog2-web-interface/src/components/inputs/InputsDotBadge.tsx b/graylog2-web-interface/src/components/inputs/InputsDotBadge.tsx index 7eecea1edef8..ebec5b468400 100644 --- a/graylog2-web-interface/src/components/inputs/InputsDotBadge.tsx +++ b/graylog2-web-interface/src/components/inputs/InputsDotBadge.tsx @@ -16,7 +16,7 @@ */ import * as React from 'react'; -import useInputsStates from 'hooks/useInputsStates'; +import useInputStateSummary from 'hooks/useInputStateSummary'; import MenuItemDotBadge from 'components/navigation/MenuItemDotBadge'; type Props = { @@ -26,18 +26,14 @@ type Props = { }; const InputsDotBadge = ({ text, hasExternalIssues = false, externalIssuesTitle = '' }: Props) => { - const { data, isLoading } = useInputsStates(); + const { hasProblematicInputs, isLoading } = useInputStateSummary(); if (isLoading) { return <>{text}; } - const hasFailedOrSetupInputs = Object.values(data).some((inputStateByNode) => - Object.values(inputStateByNode).some((node) => ['FAILED', 'FAILING', 'SETUP'].includes(node.state)), - ); - - const showDot = hasFailedOrSetupInputs || hasExternalIssues; - const title = hasFailedOrSetupInputs ? 'Some inputs are in failed state or in setup mode.' : externalIssuesTitle; + const showDot = hasProblematicInputs || hasExternalIssues; + const title = hasProblematicInputs ? 'Some inputs are in failed state or in setup mode.' : externalIssuesTitle; return ; }; diff --git a/graylog2-web-interface/src/hooks/useInputStateSummary.ts b/graylog2-web-interface/src/hooks/useInputStateSummary.ts new file mode 100644 index 000000000000..f99fcc98a53c --- /dev/null +++ b/graylog2-web-interface/src/hooks/useInputStateSummary.ts @@ -0,0 +1,50 @@ +/* + * 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 + * . + */ +import { useQuery } from '@tanstack/react-query'; + +import { SystemInputStates } from '@graylog/server-api'; + +import { defaultOnError } from 'util/conditional/onError'; + +type StateSummary = { [inputId: string]: Array }; + +const PROBLEMATIC_STATES = new Set(['FAILED', 'FAILING', 'SETUP']); + +const fetchStateSummary = (): Promise => + SystemInputStates.summary(); + +const useInputStateSummary = (): { hasProblematicInputs: boolean; isLoading: boolean } => { + const { data, isLoading } = useQuery({ + queryKey: ['inputs', 'state-summary'], + queryFn: () => + defaultOnError( + fetchStateSummary(), + 'Loading input state summary failed with status', + 'Could not load input state summary.', + ), + refetchInterval: 5000, + retry: false, + }); + + const hasProblematicInputs = data + ? Object.values(data).some((states) => states.some((s) => PROBLEMATIC_STATES.has(s))) + : false; + + return { hasProblematicInputs, isLoading }; +}; + +export default useInputStateSummary; diff --git a/graylog2-web-interface/src/routing/ApiRoutes.ts b/graylog2-web-interface/src/routing/ApiRoutes.ts index 0e02abb3760a..2b1c9797915b 100644 --- a/graylog2-web-interface/src/routing/ApiRoutes.ts +++ b/graylog2-web-interface/src/routing/ApiRoutes.ts @@ -217,6 +217,7 @@ const ApiRoutes = { references: (inputId: string) => ({ url: `/system/inputs/references/${inputId}` }), }, InputStatesController: { + summary: () => ({ url: '/system/inputstates/summary' }), start: (inputId: string) => ({ url: `/system/inputstates/${inputId}` }), stop: (inputId: string) => ({ url: `/system/inputstates/${inputId}` }), },