WIP: implement streams

.gitignore

@@ -3,3 +3,4 @@ node_modules/*
 target/*
 
 
+/target/

README.md

@@ -1,6 +1,6 @@
 # RedisLookupPlugin Plugin for Graylog
 
-Plugin to add Redis Data Adapter in read/write to graylog so you can store and retrieve key/values from pipelines  
+Plugin to add Redis Data Adapter in read/write to graylog so you can store and retrieve key/values and lists of string values from pipelines  
 Support Redis authentication (with password and username/password)  
 
 **Required Graylog version:** 5.0 and later
@@ -8,7 +8,7 @@ Support Redis authentication (with password and username/password)
 Installation
 ------------
 
-[Download the plugin](https://git.nosd.in/yo/graylog-redis-lookup-plugin/releases/download/v1.0.0/graylog-plugin-redis-lookup-1.0.0.jar)
+[Download the plugin](https://git.nosd.in/yo/graylog-redis-lookup-plugin/releases/download/v1.0.3/graylog-plugin-redis-lookup-1.0.3.jar)
 and place the `.jar` file in your Graylog plugin directory. The plugin directory
 is the `plugins/` folder relative from your `graylog-server` directory by default
 and can be configured in your `graylog.conf` file.
@@ -28,6 +28,11 @@ dramatically by making use of hot reloading. To do this, do the following:
 
 Usage
 -----
+### About data types
+* "Strings" lookup table will use GET/SET redis commands. Designed to set, get, expire, delete keys.  
+* "Streams" is designed for a "fire-and-forget" usage. Only "lookup_set_value" will work. Please note there is currently no TTL on stream, so you need to have an external process to purge your stream if you do not want it to grow indefinitely.
+
+### Usage in pipelines
 
 * Create data adapter, cache (or not), lookup table  
 * Use 'lookup_set_value(lookup_table, key, value, [ttl])' to create or update key in redis  
@@ -35,9 +40,19 @@ Usage
 * Use 'lookup_clear_key(lookup_table, key)' to remove key  
 * Use 'lookup_has_value(lookup_table, key)' to test key existence  
 * Use 'lookup_assign_ttl(lookup_table, key, ttl)' to change TTL of existing key  
+* Use 'lookup_set_string_list(lookup_table, key, value, [ttl])' to create a list named "key"  
+* Use 'lookup_add_string_list(lookup_table, key, value, [keep_duplicates])' to add value list to existing list  
+* Use 'lookup_remove_string_list(lookup_table, key, value) to remove a string from list "key"  
 
-By default keys will be created in Redis with the default TTL defined at data adapter creation time  
+By default single value keys will be created in Redis with the default TTL defined at data adapter creation time  
 
+Be aware that only setting TTL with 'lookup_assign_ttl' of 'lookup_set_string_list' alter TTL value in Redis ; so a list created with a TTL of 3600 will expire in 3600 seconds, even if it was updated with 'lookup_add_string_list' some seconds before expiration.  
+
+Known bugs
+----------
+
+* Deletion via lookup_remove_string_list, lookup_clear_key or keep_duplicates=false sometimes not done.  
+* lookup_add_string_list : Each character of a string list is considered an item  
 
 Getting started
 ---------------

dependency-reduced-pom.xml

@@ -10,7 +10,7 @@
   <groupId>in.nosd.redis</groupId>
   <artifactId>graylog-plugin-redis-lookup</artifactId>
   <name>${project.artifactId}</name>
-  <version>1.0.2</version>
+  <version>1.0.3</version>
   <description>Graylog ${project.artifactId} plugin.</description>
   <url>https://www.graylog.org</url>
   <developers>

pom.xml

@@ -31,7 +31,7 @@
 
     <groupId>in.nosd.redis</groupId>
     <artifactId>graylog-plugin-redis-lookup</artifactId>
-    <version>1.0.2</version>
+    <version>1.0.3</version>
     <packaging>jar</packaging>
 
     <name>${project.artifactId}</name>

src/main/java/in/nosd/redis/dataadapters/RedisLookupDataAdapter.java

@@ -60,20 +60,32 @@ import java.util.StringJoiner;
 
 public class RedisLookupDataAdapter extends LookupDataAdapter {    
     private static final Logger LOG = LoggerFactory.getLogger(RedisLookupDataAdapter.class);
-    
+
     public static final String NAME = "RedisLookup";
-    
+
     private static final Duration REFRESH_INTERVAL_DURATION = Duration.ZERO;
-    
+
     private final Config config;
     private final RedisClient client;
     private RedisCommands<String, String> commands;
     private StatefulRedisConnection<String, String> connection;
+
     private final Timer redisGetRequestTimer;
     private final Meter redisGetRequestErrors;
     private final Timer redisSetRequestTimer;
     private final Meter redisSetRequestErrors;
-
+    private final Timer redisDelRequestTimer;
+    private final Meter redisDelRequestErrors;
+    private final Timer redisAssignTtlRequestTimer;
+    private final Meter redisAssignTtlRequestErrors;
+    private final Timer redisAddStringListRequestTimer;
+    private final Meter redisAddStringListRequestErrors;
+    private final Timer redisSetStringListRequestTimer;
+    private final Meter redisSetStringListRequestErrors;
+    private final Timer redisSetStringListWithTtlRequestTimer;
+    private final Meter redisSetStringListWithTtlRequestErrors;
+    private final Timer redisRemoveStringListRequestTimer;
+    private final Meter redisRemoveStringListRequestErrors;
 
     @Inject
     public RedisLookupDataAdapter(@Assisted("dto") DataAdapterDto dto,
@@ -98,8 +110,23 @@ public class RedisLookupDataAdapter extends LookupDataAdapter {
         this.redisGetRequestErrors = metricRegistry.meter(MetricRegistry.name(getClass(), "redisGetRequestErrors"));
         this.redisSetRequestTimer = metricRegistry.timer(MetricRegistry.name(getClass(), "redisSetRequestTime"));
         this.redisSetRequestErrors = metricRegistry.meter(MetricRegistry.name(getClass(), "redisSetRequestErrors"));
+        this.redisDelRequestTimer = metricRegistry.timer(MetricRegistry.name(getClass(), "redisDelRequestTime"));
+        this.redisDelRequestErrors = metricRegistry.meter(MetricRegistry.name(getClass(), "redisDelRequestErrors"));
+        this.redisAssignTtlRequestTimer = metricRegistry.timer(MetricRegistry.name(getClass(), "redisAssignTtlRequestTime"));
+        this.redisAssignTtlRequestErrors = metricRegistry.meter(MetricRegistry.name(getClass(), "redisAssignTtlRequestErrors"));
+        this.redisAddStringListRequestTimer = metricRegistry.timer(MetricRegistry.name(getClass(), "redisAddStringListRequestTime"));
+        this.redisAddStringListRequestErrors = metricRegistry.meter(MetricRegistry.name(getClass(), "redisAddStringListRequestErrors"));
+        this.redisSetStringListRequestTimer = metricRegistry.timer(MetricRegistry.name(getClass(), "redisSetStringListRequestTime"));
+        this.redisSetStringListRequestErrors = metricRegistry.meter(MetricRegistry.name(getClass(), "redisSetStringListRequestErrors"));
+        this.redisSetStringListWithTtlRequestTimer = metricRegistry.timer(MetricRegistry.name(getClass(), "redisSetStringListWithTtlRequestTime"));
+        this.redisSetStringListWithTtlRequestErrors = metricRegistry.meter(MetricRegistry.name(getClass(), "redisSetStringListWithTtlRequestErrors"));
+        this.redisRemoveStringListRequestTimer = metricRegistry.timer(MetricRegistry.name(getClass(), "redisRemoveStringListRequestTime"));
+        this.redisRemoveStringListRequestErrors = metricRegistry.meter(MetricRegistry.name(getClass(), "redisRemoveStringListRequestErrors"));
     }
 
+/**************************************************************************
+* LookupDataAdapter functions
+**************************************************************************/
     @Override
     protected void doStart() throws Exception {        
         connection = this.client.connect();
@@ -133,15 +160,22 @@ public class RedisLookupDataAdapter extends LookupDataAdapter {
             return getEmptyResult();
         }
         try {
-            final String value = this.commands.get(trimmedKey);
-            if (value == null) {
-                LOG.debug("Redis GET request for key <{}> returned null, key do not exists.", trimmedKey);
-                redisGetRequestErrors.mark();
-                return LookupResult.empty();
+            // Get item type and existence
+            final String type = this.commands.type(trimmedKey);
+            switch(type) {
+                case "none":
+                    LOG.debug("Redis TYPE request for key <{}> returned null, key do not exists.", trimmedKey);
+                    redisGetRequestErrors.mark();
+                    return LookupResult.empty();
+                case "list":
+                    final List<String> result = this.commands.lrange(trimmedKey, 0, -1);
+                    return LookupResult.withoutTTL().stringListValue(result).build();
+                default:
+                    final String value = getSingleValue(trimmedKey);
+                    return LookupResult.single(value);
             }
-            return LookupResult.single(value);
         } catch (Exception e) {
-            LOG.error("Redis GET request error for key <{}>", trimmedKey, e);
+            LOG.error("Exception: Redis GET request error for key <{}>", trimmedKey, e);
             redisGetRequestErrors.mark();
             return LookupResult.empty();
         } finally {
@@ -177,7 +211,7 @@ public class RedisLookupDataAdapter extends LookupDataAdapter {
             }
             return LookupResult.single(value.toString());
         } catch (Exception e) {
-            LOG.error("Redis SET(EX) key <{}> to value <{}> with TTL <{}> returned an exception: {}", key, value, ttlSec, e);
+            LOG.error("Exception: Redis SET(EX) key <{}> to value <{}> with TTL <{}> returned {}", key, value, ttlSec, e);
             redisSetRequestErrors.mark();
             return LookupResult.withError();
         } finally {
@@ -187,57 +221,64 @@ public class RedisLookupDataAdapter extends LookupDataAdapter {
 
     @Override
     public void clearKey(Object key) {
-        final Timer.Context time = redisSetRequestTimer.time();
+        final Timer.Context time = redisDelRequestTimer.time();
         final String trimmedKey = StringUtils.trimToNull(key.toString());
         try {
-            final Long result = this.commands.del(key.toString());
+            final Long result = this.commands.del(trimmedKey);
             if (result != 1) {
-                LOG.debug("Redis DEL key <{}> returned {}", key, result);
-                redisSetRequestErrors.mark();
+                LOG.debug("Redis DEL key <{}> returned {}", trimmedKey, result);
+                redisDelRequestErrors.mark();
             }
             return;
         } catch (Exception e) {
-            LOG.error("Redis DEL key <{}> returned {}", key, e);
-            redisSetRequestErrors.mark();
+            LOG.error("Exception: Redis DEL key <{}> returned {}", trimmedKey, e);
+            redisDelRequestErrors.mark();
             return;
         } finally {
             time.stop();
         }
     }
 
+    private String getSingleValue(String key) {
+        final String value = this.commands.get(key);
+        return value;
+    }
+
     private LookupResult setExpire(String key, Long ttl) {
         try {
-            final Boolean result = this.commands.expire(key, ttl);
-            if (!result) {
-                LOG.warn("Redis EXPIRE key <{}> to <{}> returned {}", key, ttl, result);
+            if (!this.commands.expire(key, ttl)) {
+                LOG.warn("Redis EXPIRE key <{}> to <{}> returned false (key does not exist or the timeout could not be set)", key, ttl);
                 return LookupResult.withError();
             }
-            final String value = this.commands.get(key);
-            return LookupResult.single(value.toString());
-        } catch (Exception e) {
-            LOG.error("Redis EXPIRE key <{}> to <{}> returned {}", key, ttl, e);
-            return LookupResult.withError(e.toString());
+            return LookupResult.single(this.commands.get(key).toString());
         }
+        catch (Exception e) {
+            // lettuce 6.3.0 returns "WRONGTYPE Operation against a key holding the wrong kind of value" when EXPIRE on a list key, but do the job.
+            if (e.getMessage().startsWith("WRONGTYPE Operation against a key holding the wrong kind of value")) {
+            } else {
+                LOG.error("Exception: Redis EXPIRE key <{}> to <{}> returned {}", key, ttl, e);
+                return LookupResult.withError(e.toString());
+            }
+        }
+        return LookupResult.single(this.commands.get(key).toString());
     }
 
     private LookupResult setPersist(String key) {
         try {
-            final Boolean result = this.commands.persist(key);
-            if (!result) {
-                LOG.warn("Redis PERSIST key <{}> returned {}", key, result);
+            if (!this.commands.persist(key)) {
+                LOG.debug("Redis PERSIST key <{}> returned false (key does not exist or does not have an associated timeout)", key);
                 return LookupResult.withError();
             }
-            final String value = this.commands.get(key);
-            return LookupResult.single(value.toString());
+            return LookupResult.single(this.commands.get(key).toString());
         } catch (Exception e) {
-            LOG.error("Redis PERSIST key <{}> returned {}", key, e);
+            LOG.error("Exception: Redis PERSIST key <{}> returned {}", key, e);
             return LookupResult.withError(e.toString());
         }
     }
 
     // TTL -1 = never expire
     public LookupResult assignTtl(Object key, Long ttlSec) {
-        final Timer.Context time = redisSetRequestTimer.time();
+        final Timer.Context time = redisAssignTtlRequestTimer.time();
         final String trimmedKey = StringUtils.trimToNull(key.toString());
         try {
             if (ttlSec > 0) {
@@ -246,9 +287,44 @@ public class RedisLookupDataAdapter extends LookupDataAdapter {
                 return setPersist(trimmedKey);
             }
         } catch (Exception e) {
-            LOG.error("assignTtl <{}> to key <{}> returned {}", ttlSec, trimmedKey, e);
-            redisSetRequestErrors.mark();
-            return LookupResult.withError();
+            // lettuce 6.3.0 returns "WRONGTYPE Operation against a key holding the wrong kind of value" when TTL on a list key, but do the job.
+            if (e.getMessage().startsWith("WRONGTYPE Operation against a key holding the wrong kind of value")) {
+            } else {
+                LOG.error("Exception: assignTtl <{}> to key <{}> returned {}", ttlSec, trimmedKey, e);
+                redisAssignTtlRequestErrors.mark();
+                return LookupResult.withError();
+            }
+        } finally {
+            time.stop();
+        }
+        return LookupResult.single(this.commands.get(trimmedKey));
+    }
+
+    @Override
+    public LookupResult addStringList(Object key, List<String> listValue, boolean keepDuplicates) {
+        final Timer.Context time = redisAddStringListRequestTimer.time();
+        final String trimmedKey = StringUtils.trimToNull(key.toString());
+        if (trimmedKey == null) {
+            LOG.debug("A blank key was supplied");
+            return getEmptyResult();
+        }
+        try {
+            if (!keepDuplicates) {
+                removeStringList(trimmedKey, listValue);
+            }
+            
+            LOG.info("Redis addStringList: List is <{}> items", listValue.size());
+            LOG.info("Redis addStringList: List is <{}>", listValue);
+            
+            final Long len = this.commands.rpush(trimmedKey, listValue.toArray(new String[0]));
+            if (len > 0) {
+                return LookupResult.withoutTTL().stringListValue(this.commands.lrange(trimmedKey, 0, -1)).build();
+            }
+            return LookupResult.empty();
+        } catch (Exception e) {
+            LOG.error("Exception: Redis RPUSH request error for key <{}>: <{}>", trimmedKey, e);
+            redisAddStringListRequestErrors.mark();
+            return LookupResult.empty();
         } finally {
             time.stop();
         }
@@ -256,7 +332,70 @@ public class RedisLookupDataAdapter extends LookupDataAdapter {
 
     @Override
     public LookupResult setStringList(Object key, List<String> listValue) {
-        return LookupResult.empty();
+        final Timer.Context time = redisSetStringListRequestTimer.time();
+        final String trimmedKey = StringUtils.trimToNull(key.toString());
+        if (trimmedKey == null) {
+            LOG.debug("A blank key was supplied");
+            return getEmptyResult();
+        }
+        try {
+            // We need to replace list, so we delete it first
+            this.commands.ltrim(trimmedKey, 1, 0);
+            final Long len = this.commands.rpush(trimmedKey, listValue.toArray(new String[0]));
+            if (len > 0) {
+                return LookupResult.withoutTTL().stringListValue(this.commands.lrange(trimmedKey, 0, -1)).build();
+            }
+            return LookupResult.empty();
+        } catch (Exception e) {
+            LOG.error("Exception: Redis RPUSH request error for key <{}>: <{}>", trimmedKey, e);
+            redisSetStringListRequestErrors.mark();
+            return LookupResult.empty();
+        } finally {
+            time.stop();
+        }
+    }
+
+    public LookupResult setStringListWithTtl(Object key, List<String> listValue, Long ttlSec) {
+        final Timer.Context time = redisSetStringListWithTtlRequestTimer.time();
+        try {
+            setStringList(key, listValue);
+            return(assignTtl(key, ttlSec));
+        } catch (Exception e) {
+            // This exception comes from assignTtl
+            if (e.getMessage().startsWith("WRONGTYPE Operation against a key holding the wrong kind of value")) {
+            } else {
+                LOG.error("Exception: Redis RPUSH request error for key <{}>: <{}>", key, e);
+                redisSetStringListWithTtlRequestErrors.mark();
+                return LookupResult.empty();
+            }
+        } finally {
+            time.stop();
+        }
+        return LookupResult.withoutTTL().stringListValue(this.commands.lrange(StringUtils.trimToNull(key.toString()), 0, -1)).build();
+    }
+
+    @Override
+    public LookupResult removeStringList(Object key, List<String> listValue) {
+        final Timer.Context time = redisRemoveStringListRequestTimer.time();
+        final String trimmedKey = StringUtils.trimToNull(key.toString());
+        if (trimmedKey == null) {
+            LOG.debug("A blank key was supplied");
+            return getEmptyResult();
+        }
+        try {
+            long removed = 0;
+            for (String value : listValue) {
+                removed += this.commands.lrem(trimmedKey, 0, value);
+            }
+            LOG.debug("Redis LREM for key <{}> and value <{}> deleted <{}> items", trimmedKey, listValue, removed);
+            return LookupResult.withoutTTL().stringListValue(this.commands.lrange(trimmedKey, 0, -1)).build();
+        } catch (Exception e) {
+            LOG.error("Exception: Redis LREM request error for key <{}> and value <{}>: <{}>", trimmedKey, listValue, e);
+            redisRemoveStringListRequestErrors.mark();
+            return LookupResult.empty();
+        } finally {
+            time.stop();
+        }
     }
 
     public interface Factory extends LookupDataAdapter.Factory2<RedisLookupDataAdapter> {
@@ -279,6 +418,7 @@ public class RedisLookupDataAdapter extends LookupDataAdapter {
                     .redisHost("127.0.0.1")
                     .redisPort(6379)
                     .redisDB(0)
+                    .redisType("strings")
                     .redisKeyTTL(-1)
                     .redisUsername("")
                     .redisPassword("")
@@ -310,6 +450,10 @@ public class RedisLookupDataAdapter extends LookupDataAdapter {
         @JsonProperty("redis_database")
         @Min(0)
         public abstract int redisDB();
+        
+        @JsonProperty("redis_type")
+        @NotEmpty
+        public abstract String redisType();
 
         @JsonProperty("redis_ttl")
         @Min(0)
@@ -357,6 +501,9 @@ public class RedisLookupDataAdapter extends LookupDataAdapter {
 
             @JsonProperty("redis_database")
             public abstract Builder redisDB(int redisDB);
+            
+            @JsonProperty("redis_type")
+            public abstract Builder redisType(String redisType);
 
             @JsonProperty("redis_ttl")
             public abstract Builder redisKeyTTL(long redisKeyTTL);

src/web/dataadapters/RedisLookupAdapterFieldSet.jsx

@@ -16,13 +16,23 @@
  */
 import React, { useCallback, useEffect, useRef, useState } from 'react';
 import PropTypes from 'prop-types';
-//import { Button } from 'components/graylog';
+
+import ObjectUtils from 'util/ObjectUtils';
 import { Input } from 'components/bootstrap';
+// "Missing or invalid plugin" in "dataAdapter create/Data adapter type" list when imported
+//import { Select } from './components/common';
+//import { Select } from '../../../../graylog2-server/graylog2-web-interface/src/components/common';
 
 class RedisLookupAdapterFieldSet extends React.Component {
   static propTypes = {
     config: PropTypes.shape({
       redis_host: PropTypes.string.isRequired,
+      redis_port: PropTypes.number.isRequired,
+      redis_database: PropTypes.number.isRequired,
+      redis_type: PropTypes.string.isRequired,
+      redis_ttl: PropTypes.number.isRequired,
+      redis_username: PropTypes.string,
+      redis_password: PropTypes.string,
     }).isRequired,
     updateConfig: PropTypes.func.isRequired,
     handleFormEvent: PropTypes.func.isRequired,
@@ -30,14 +40,14 @@ class RedisLookupAdapterFieldSet extends React.Component {
     validationMessage: PropTypes.func.isRequired,
   };
   
-  handleSelect = (fieldName) => {
-    return (selectedIndicator) => {
-      const config = lodash.cloneDeep(this.props.config);
-      config[fieldName] = selectedIndicator;
-      this.props.updateConfig(config);
-    };
+/*  _onRedisTypeSelect = (type) => {
+    const { config, updateConfig } = this.props;
+    const newConfig = ObjectUtils.clone(config);
+    
+    newConfig.redis_type = type;
+    updateConfig(newConfig);
   };
-
+*/
   render() {
     const { config } = this.props;
  
@@ -76,6 +86,18 @@ class RedisLookupAdapterFieldSet extends React.Component {
         value={config.redis_database}
         labelClassName="col-sm-3"
         wrapperClassName="col-sm-9" />
+      <Input type="text"
+        id="redis_type"
+        name="redis_type"
+        label="Redis data type"
+        required
+        onChange={this.props.handleFormEvent}
+        help={this.props.validationMessage('redis_type', 'Redis data type used for lookups. Should be one of "strings", "streams".')}
+        bsStyle={this.props.validationState('redis_type')}
+        value={config.redis_type}
+        labelClassName="col-sm-3"
+        wrapperClassName="col-sm-9">
+      </Input>
       <Input type="text"
         id="redis_ttl"
         name="redis_ttl"

src/web/dataadapters/RedisLookupAdapterSummary.jsx

@@ -34,6 +34,16 @@ class RedisLookupAdapterSummary extends React.Component {
       <dl>
       <dt>Redis host</dt>
       <dd>{config.redis_host || 'n/a'}</dd>
+      <dt>Redis port</dt>
+      <dd>{config.redis_port || 'n/a'}</dd>
+      <dt>Redis database</dt>
+      <dd>{config.redis_database}</dd>
+      <dt>Redis key TTL</dt>
+      <dd>{config.redis_ttl || 'n/a'}</dd>
+      <dt>Redis username</dt>
+      <dd>{config.redis_username || 'n/a'}</dd>
+      <dt>Redis password</dt>
+      <dd>******</dd>
       </dl>
     );
   }