add telemetry input and dwh traffic metrics

graylog2-server/src/main/java/org/graylog2/bindings/PeriodicalBindings.java

@@ -35,7 +35,7 @@
 import org.graylog2.periodical.LeaderPresenceCheckPeriodical;
 import org.graylog2.periodical.NodePingThread;
 import org.graylog2.periodical.ThrottleStateUpdaterThread;
-import org.graylog2.periodical.TrafficCounterCalculator;
+import org.graylog2.periodical.TrafficCounterPeriodical;
 import org.graylog2.periodical.UserSessionTerminationPeriodical;
 import org.graylog2.periodical.VersionCheckThread;
 import org.graylog2.plugin.periodical.Periodical;
@@ -58,7 +58,7 @@ protected void configure() {
         periodicalBinder.addBinding().to(ClusterEventPeriodical.class);
         periodicalBinder.addBinding().to(ClusterEventCleanupPeriodical.class);
         periodicalBinder.addBinding().to(IndexRangesCleanupPeriodical.class);
-        periodicalBinder.addBinding().to(TrafficCounterCalculator.class);
+        periodicalBinder.addBinding().to(TrafficCounterPeriodical.class);
         periodicalBinder.addBinding().to(IndexFieldTypePollerPeriodical.class).asEagerSingleton();
         periodicalBinder.addBinding().to(ScheduleTriggerCleanUp.class);
         periodicalBinder.addBinding().to(ESVersionCheckPeriodical.class);

graylog2-server/src/main/java/org/graylog2/bindings/ServerBindings.java

@@ -100,6 +100,8 @@
 import org.graylog2.system.jobs.SystemJobManager;
 import org.graylog2.system.shutdown.GracefulShutdown;
 import org.graylog2.system.stats.ClusterStatsModule;
+import org.graylog2.system.traffic.OpenTrafficCounterCalculator;
+import org.graylog2.system.traffic.TrafficCounterCalculator;
 import org.graylog2.system.traffic.TrafficCounterService;
 import org.graylog2.system.traffic.TrafficUpdater;
 import org.graylog2.telemetry.enterprise.DefaultTelemetryEnterpriseDataProvider;
@@ -215,6 +217,7 @@ private void bindInterfaces() {
 
         OptionalBinder.newOptionalBinder(binder(), DataTieringStatusService.class).setDefault().to(DefaultDataTieringStatusService.class);
 
+        Multibinder.newSetBinder(binder(), TrafficCounterCalculator.class).addBinding().to(OpenTrafficCounterCalculator.class);
         OptionalBinder.newOptionalBinder(binder(), TrafficUpdater.class).setDefault().to(TrafficCounterService.class).asEagerSingleton();
     }
 

graylog2-server/src/main/java/org/graylog2/periodical/TrafficCounterPeriodical.java

@@ -0,0 +1,90 @@
+/*
+ * 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.graylog2.periodical;
+
+import jakarta.inject.Inject;
+import org.graylog2.plugin.Tools;
+import org.graylog2.plugin.periodical.Periodical;
+import org.graylog2.system.traffic.TrafficCounterCalculator;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Set;
+
+public class TrafficCounterPeriodical extends Periodical {
+    private static final Logger LOG = LoggerFactory.getLogger(TrafficCounterPeriodical.class);
+    private final Set<TrafficCounterCalculator> trafficCounterCalculators;
+
+    @Inject
+    public TrafficCounterPeriodical(Set<TrafficCounterCalculator> trafficCounterCalculators) {
+        this.trafficCounterCalculators = trafficCounterCalculators;
+    }
+
+    @Override
+    public void doRun() {
+        final DateTime now = Tools.nowUTC();
+        final int secondOfMinute = now.getSecondOfMinute();
+        // on the top of every minute, we calculate the traffic
+        if (secondOfMinute == 0) {
+            trafficCounterCalculators.forEach(trafficCounterCalculator -> {
+                DateTime previousMinute = now.minusMinutes(1);
+                trafficCounterCalculator.calculate(previousMinute);
+            });
+        }
+    }
+
+    @Override
+    public boolean runsForever() {
+        return false;
+    }
+
+    @Override
+    public boolean stopOnGracefulShutdown() {
+        return true;
+    }
+
+    @Override
+    public boolean leaderOnly() {
+        return false;
+    }
+
+    @Override
+    public boolean startOnThisNode() {
+        return true;
+    }
+
+    @Override
+    public boolean isDaemon() {
+        return true;
+    }
+
+    @Override
+    public int getInitialDelaySeconds() {
+        return 0;
+    }
+
+    @Override
+    public int getPeriodSeconds() {
+        return 1;
+    }
+
+    @Override
+    protected Logger getLogger() {
+        return LOG;
+    }
+}

graylog2-server/src/main/java/org/graylog2/system/traffic/OpenTrafficCounterCalculator.java

@@ -0,0 +1,79 @@
+/*
+ * 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.graylog2.system.traffic;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.MetricRegistry;
+import com.github.joschi.jadconfig.util.Size;
+import jakarta.inject.Inject;
+import org.graylog2.plugin.GlobalMetricNames;
+import org.graylog2.plugin.system.NodeId;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class OpenTrafficCounterCalculator implements TrafficCounterCalculator {
+    private static final Logger LOG = LoggerFactory.getLogger(OpenTrafficCounterCalculator.class);
+    private final NodeId nodeId;
+    private final TrafficUpdater trafficUpdater;
+
+    private volatile long previousInputBytes = 0L;
+    private volatile long previousOutputBytes = 0L;
+    private volatile long previousDecodedBytes = 0L;
+    private final Counter inputCounter;
+    private final Counter outputCounter;
+    private final Counter decodedCounter;
+
+    @Inject
+    public OpenTrafficCounterCalculator(NodeId nodeId, TrafficUpdater trafficUpdater, MetricRegistry metricRegistry) {
+        this.nodeId = nodeId;
+        this.trafficUpdater = trafficUpdater;
+        inputCounter = metricRegistry.counter(GlobalMetricNames.INPUT_TRAFFIC);
+        outputCounter = metricRegistry.counter(GlobalMetricNames.OUTPUT_TRAFFIC);
+        decodedCounter = metricRegistry.counter(GlobalMetricNames.DECODED_TRAFFIC);
+    }
+
+
+    @Override
+    public void calculate(DateTime previousMinute) {
+        LOG.trace("Calculating input and output traffic for the previous minute");
+
+        final long currentInputBytes = inputCounter.getCount();
+        final long currentOutputBytes = outputCounter.getCount();
+        final long currentDecodedBytes = decodedCounter.getCount();
+
+        final long inputLastMinute = currentInputBytes - previousInputBytes;
+        previousInputBytes = currentInputBytes;
+        final long outputBytesLastMinute = currentOutputBytes - previousOutputBytes;
+        previousOutputBytes = currentOutputBytes;
+        final long decodedBytesLastMinute = currentDecodedBytes - previousDecodedBytes;
+        previousDecodedBytes = currentDecodedBytes;
+
+        if (LOG.isDebugEnabled()) {
+            final Size in = Size.bytes(inputLastMinute);
+            final Size out = Size.bytes(outputBytesLastMinute);
+            final Size decoded = Size.bytes(decodedBytesLastMinute);
+            LOG.debug("Traffic in the last minute: in: {} bytes ({} MB), out: {} bytes ({} MB}), decoded: {} bytes ({} MB})",
+                    in, in.toMegabytes(), out, out.toMegabytes(), decoded, decoded.toMegabytes());
+        }
+        trafficUpdater.updateTraffic(previousMinute,
+                nodeId,
+                inputLastMinute,
+                outputBytesLastMinute,
+                decodedBytesLastMinute);
+    }
+}

graylog2-server/src/main/java/org/graylog2/system/traffic/TrafficCounterCalculator.java

@@ -0,0 +1,24 @@
+/*
+ * 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.graylog2.system.traffic;
+
+import org.joda.time.DateTime;
+
+public interface TrafficCounterCalculator {
+
+    void calculate(DateTime previousMinute);
+}