Merge pull request #2460 from telefonicaid/task/alarm_channel_full

log warn when channel is near full

Author: fgalan

CHANGES_NEXT_RELEASE

@@ -1,3 +1,4 @@
+- [cygnus-ngsi] Log warn when channel is near to be full (>90%) (#2459)
 - [cygnus-common] Upgrade postgresql from 42.7.2 to 42.7.5
 - [cygnus-ngsi] Upgrade Debian version from 12.9 to 12.10 in Dockerfile
 

cygnus-common/src/main/java/com/telefonica/iot/cygnus/channels/CygnusChannel.java

@@ -60,7 +60,13 @@ public interface CygnusChannel {
      * @return The number of take operations on the channel that failed
      */
     long getNumTakesFail();
-    
+
+    /**
+     * Gets the number of usage percent of the channel.
+     * @return The number of usage percente of the channel
+     */
+    double getUsage();
+
     /**
      * Sets the number of put operations on the channel that went OK.
      * @param n The number to be set

cygnus-common/src/main/java/com/telefonica/iot/cygnus/channels/CygnusFileChannel.java

@@ -115,5 +115,13 @@ public void setNumTakesOK(long n) {
     public void setNumTakesFail(long n) {
         accTakesFail = channelCounterRef.getEventTakeAttemptCount() - channelCounterRef.getEventTakeSuccessCount();
     } // setNumTakesFail
+
+    @Override
+    public double getUsage() {
+        long currentSize = channelCounterRef.getChannelSize();
+        long maxCapacity = channelCounterRef.getChannelCapacity();
+        double usagePercentage = (double) currentSize / maxCapacity * 100;
+        return usagePercentage;
+    } // getUsage
 
 } // CygnusFileChannel

cygnus-common/src/main/java/com/telefonica/iot/cygnus/channels/CygnusMemoryChannel.java

@@ -120,4 +120,13 @@ public void setNumTakesFail(long n) {
         accTakesFail = channelCounterRef.getEventTakeAttemptCount() - channelCounterRef.getEventTakeSuccessCount();
     } // setNumTakesFail
 
+    @Override
+    public double getUsage() {
+        long currentSize = channelCounterRef.getChannelSize();
+        long maxCapacity = channelCounterRef.getChannelCapacity();
+        double usagePercentage = (double) currentSize / maxCapacity * 100;
+        return usagePercentage;
+    } // getUsage
+
 } // CygnusMemoryChannel
+

cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSISink.java

@@ -31,6 +31,7 @@
 import com.telefonica.iot.cygnus.interceptors.NGSIEvent;
 import com.telefonica.iot.cygnus.log.CygnusLogger;
 import com.telefonica.iot.cygnus.sinks.Enums.DataModel;
+import com.telefonica.iot.cygnus.channels.CygnusChannel;
 import static com.telefonica.iot.cygnus.sinks.Enums.DataModel.DMBYATTRIBUTE;
 import static com.telefonica.iot.cygnus.sinks.Enums.DataModel.DMBYENTITY;
 import static com.telefonica.iot.cygnus.sinks.Enums.DataModel.DMBYENTITYTYPE;
@@ -58,6 +59,7 @@
 import org.apache.flume.Sink.Status;
 import org.apache.flume.Transaction;
 import org.apache.flume.ChannelException;
+import org.apache.flume.ChannelFullException;
 import org.apache.flume.conf.Configurable;
 import org.apache.logging.log4j.ThreadContext;
 
@@ -500,15 +502,25 @@ protected void doRollbackAgain(Accumulator rollbackedAccumulation) {
         } // if else
     } // doRollbackAgain
 
+
     private Status processNewBatches() {
         // Get the channel
         Channel ch = getChannel();
+
+        double channelUsage = 0;
 
         // Start a Flume transaction (it is not the same than a Cygnus transaction!)
         Transaction txn = ch.getTransaction();
         try {
             txn.begin();
 
+            CygnusChannel cygnusch = (CygnusChannel) ch;
+            channelUsage = cygnusch.getUsage();
+            LOGGER.debug("Channel usage: " + channelUsage + "% in sink " + this.getName());
+            if (channelUsage > NGSIConstants.HIGH_CHANNEL_PERCENT_USAGE) {
+                LOGGER.warn("High Channel usage: " + channelUsage + "% in sink " + this.getName());
+            }
+
             // Get and process as many events as the batch size
             int currentIndex;
 

cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/utils/NGSIConstants.java

@@ -53,6 +53,8 @@ private NGSIConstants() {
 
     public static final int ROLLBACK_CHANNEL_EXCEPTION_THRESHOLD = 10;
     public static final int ROLLBACK_EXCEPTION_THRESHOLD = 10;
+
+    public static final int HIGH_CHANNEL_PERCENT_USAGE = 90;
 
     // FIWARE service and FIWARE service path specific constants
     public static final int SERVICE_HEADER_MAX_LEN      = 50;