Merge pull request #22 from Axway-API-Management-Plus/memchached

Adding Memchached to the Logstash pipeline

Author: Chris Wiechmann

.env

@@ -49,6 +49,13 @@ API_MANAGER_PASSWORD=changeme
 # parameter.
 LOGSTASH=logstash:5044
 
+# ----------------------------------------------------------------------------------------------
+# With this parameter you tell the Logstash processing pipeline which memchached to use. It is 
+# used to cache the API-Details that has been looked up from the API-Manager via the API-Builder.
+# Even if the API-Builder is already caching the result, this improves ingest performance. 
+# The default parameter works when using the docker-compose.yml
+MEMCACHED=memcached:11211
+
 # ----------------------------------------------------------------------------------------------
 # This is an optional parameter used by Filebeat to set a proper name. This allows for instance 
 # to identify the different Filebeat instances in the Kibana-Stack Monitoring dashboards. 

.github/workflows/logstash.yml

@@ -13,6 +13,10 @@ jobs:
           - 8080:8080
         env:
           MOCK_LOOKUP_API: true
+      memcached:
+        image: memcached:1.6.6-alpine
+        ports: 
+          - 11211:11211
 
     steps:
     - uses: actions/checkout@v2
@@ -30,6 +34,8 @@ jobs:
     - name: Test Logstash pipeline
       run: |
         echo Using API_BUILDER_URL: $API_BUILDER_URL
-        ./logstash-filter-verifier --diff-command="diff -y" --keep-env=API_BUILDER_URL ./logstash/test ./logstash/pipeline
+        echo Using MEMCACHED: $MEMCACHED
+        ./logstash-filter-verifier --diff-command="diff -y" --keep-env=API_BUILDER_URL --keep-env=MEMCACHED ./logstash/test ./logstash/pipeline
       env:
         API_BUILDER_URL: 'http://localhost:8080'
+        MEMCACHED: 'localhost:11211'

docker-compose.yml

@@ -79,6 +79,7 @@ services:
       - PIPELINE_WORKERS=1
       - XPACK_MONITORING_ELASTICSEARCH_HOSTS=["elasticsearch1:9200"]
       - API_BUILDER_URL=${API_BUILDER_URL}
+      - MEMCACHED=${MEMCACHED}
     ports:
       - 5044:5044
     volumes:
@@ -88,6 +89,15 @@ services:
     depends_on:
       - elasticsearch1
       - elk-traffic-monitor-api
+      - memcached
+    networks:
+      - elastic
+
+  # Memcached is used by the Logstash pipeline to cache API-Lookup information that are used to get the API-Organization
+  memcached:
+    image: memcached:1.6.6-alpine
+    links:
+      - elasticsearch1
     networks:
       - elastic
 

logstash/pipeline/pipeline.conf

@@ -64,33 +64,52 @@ filter {
        remove_field => "[transactionSummary][serviceContexts]"
        rename => ["[transactionSummary][serviceContext][org]", "[transactionSummary][serviceContext][appOrg]" ]
       }
-      # Future plan is to perform side-calls only if not yet already cached
-      #mutate {
-      #  add_field => { "apiCacheKey" => "%{[transactionSummary][serviceContext][service]}---%{[transactionSummary][path]}" }
-      #}
-      #memcached {
-      #  namespace => "api_details"
-      #  ttl => 300
-      #  get => { "apiCacheKey" => "[transactionSummary][serviceContext][serviceOrg]" }
-      #  remove_field => [ "apiCacheKey" ]
-      #}
-      #if ![transactionSummary][serviceContext][serviceOrg] {
-      # Perform a side-call to load organization details for the API
+      # If we have a ServiceConext we have to enrich it with the API-Details
       if([transactionSummary][serviceContext]) {
-        http {
-          url => "${API_BUILDER_URL}/api/elk/v1/api/lookup/api"
-          query => {
-            "apiName" => "%{[transactionSummary][serviceContext][service]}"
-            "apiPath" => "%{[transactionSummary][path]}"
+          # Create a key for the API
+        mutate {
+          add_field => { "apiCacheKey" => "%{[transactionSummary][path]}" }
+        }
+        # Lookup the cache with the created API-Key (API-Name---API-Path)
+        memcached {
+          hosts => "${MEMCACHED}"
+          namespace => "api_details"
+          get => { "%{apiCacheKey}" => "[apiDetails]" }
+        }
+        # If we have nothing in the cache, perform the Lookup via the HTTP-Builder API
+        if !([apiDetails]) {
+          http {
+            url => "${API_BUILDER_URL}/api/elk/v1/api/lookup/api"
+            query => {
+              "apiName" => "%{[transactionSummary][serviceContext][service]}"
+              "apiPath" => "%{[transactionSummary][path]}"
+            }
+            target_body => "apiDetails"
+            add_field => { "[@metadata][updateAPICache]" => "true" }
           }
-          target_body => "apiDetails"
-          add_field => { 
-            "[transactionSummary][serviceContext][apiOrg]" => "%{[apiDetails][organizationName]}"
-            "[transactionSummary][serviceContext][apiVersion]" => "%{[apiDetails][version]}" 
-            "[transactionSummary][serviceContext][apiDeprecated]" => "%{[apiDetails][deprecated]}" 
-            "[transactionSummary][serviceContext][apiState]" => "%{[apiDetails][state]}"
+        }
+        # At this point we should have the apiDetails either from the cache or looked up via HTTP
+        if([apiDetails]) {
+          mutate {
+            add_field => { 
+              "[transactionSummary][serviceContext][apiOrg]" => "%{[apiDetails][organizationName]}"
+              "[transactionSummary][serviceContext][apiVersion]" => "%{[apiDetails][version]}" 
+              "[transactionSummary][serviceContext][apiDeprecated]" => "%{[apiDetails][deprecated]}" 
+              "[transactionSummary][serviceContext][apiState]" => "%{[apiDetails][state]}"
+            }
           }
-          remove_field => [ "apiDetails", "headers" ]
+          # If the API has been looked up add it to the cache
+          if([@metadata][updateAPICache]=="true") {
+            memcached {
+              hosts => "${MEMCACHED}"
+              namespace => "api_details"
+              ttl => 3600
+              set => { "[apiDetails]" => "%{apiCacheKey}" }
+            }
+          }
+        }
+        mutate {
+          remove_field => [ "apiDetails", "headers", "updateAPICache", "apiCacheKey" ]
         }
         # If the API-Lookup failed - Clone the event which is send to an Error index and shown in Traffic-Monitor
         if("_httprequestfailure" in [tags]) {
@@ -111,7 +130,6 @@ filter {
           }
         }
       }
-      #}
     }
   } else if([logtype] == "trace") {
     if [message] =~ /^#/ {