diff --git a/extra/docker/elk/docker-compose.yml b/extra/docker/elk/docker-compose.yml
index 3c5652a..7ac58e8 100644
--- a/extra/docker/elk/docker-compose.yml
+++ b/extra/docker/elk/docker-compose.yml
@@ -15,11 +15,13 @@
       - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
 
   logstash:
-    image: docker.elastic.co/logstash/logstash-oss:6.1.3
+    image: onap/clamp-dashboard-logstash:latest
+    build:
+      context: ../../../src/main/docker/logstash
+      dockerfile: Dockerfile.logstash
     volumes:
-      - ./logstash-conf:/config-dir
+      - ../../../src/main/docker/logstash/pipeline:/usr/share/logstash/pipeline
       - ./logstash-input:/log-input
-    command: logstash -f /config-dir/logstash.conf
     depends_on:
       - elasticsearch
     networks:
@@ -28,12 +30,12 @@
       - dmaap_base_url=http://dmaapmocker:3904/
       - dmaap_user=user
       - dmaap_password=password
-      - dmaap_consumer_group=CG42
-      - dmaap_consumer_id=C42
-      - event_topic=EVENT_TOPIC
-      - notification_topic=NOTIFICATION_TOPIC
-      - request_topic=REQUEST_TOPIC
-      - elasticsearch_hosts=elasticsearch
+      - dmaap_consumer_group=clampdashboard
+      - dmaap_consumer_id=clampdashboard
+      - event_topic=DCAE-CL-EVENT
+      - notification_topic=POLICY-CL-MGT
+      - request_topic=APPC-CL
+      - elasticsearch_base_url=elasticsearch
 
   kibana:
     image: onap/clamp-dashboard-kibana:latest
diff --git a/extra/docker/elk/logstash-conf/logstash.conf b/extra/docker/elk/logstash-conf/logstash.conf
deleted file mode 100644
index 2b71686..0000000
--- a/extra/docker/elk/logstash-conf/logstash.conf
+++ /dev/null
@@ -1,148 +0,0 @@
-input {
-  http_poller {
-        urls => {
-            event_queue => {
-                method => get
-                url => "${dmaap_base_url}/events/${event_topic}/${dmaap_consumer_group}/${dmaap_consumer_id}?timeout=15000"
-                headers => {
-                    Accept => "application/json"
-                }
-                add_field => { "topic" => "${event_topic}" }
-                type => "dmaap_event"
-            }
-            notification_queue => {
-                method => get
-                url => "${dmaap_base_url}/events/${notification_topic}/${dmaap_consumer_group}/${dmaap_consumer_id}?timeout=15000"
-                headers => {
-                    Accept => "application/json"
-                }
-                add_field => { "topic" => "${notification_topic}" }
-                type => "dmaap_notification"
-            }
-            request_queue => {
-                method => get
-                url => "${dmaap_base_url}/events/${request_topic}/${dmaap_consumer_group}/${dmaap_consumer_id}?timeout=15000"
-                headers => {
-                    Accept => "application/json"
-                }
-                add_field => { "topic" => "${request_topic}" }
-                type => "dmaap_request"
-            }
-        }
-        socket_timeout => 30
-        request_timeout => 30
-        codec => "plain"
-        schedule => { "every" => "1m"  }
-  }
-}
-
-input {
-  file {
-    path => [
-      "/log-input/dmaap_evt.log"
-    ]
-    type => "dmaap_log"
-    codec => "json"
-  }
-}
-
-filter {
-
-    # parse json, split  the list into multiple events, and parse each event
-    if [type] != "dmaap_log" {
-	    # avoid noise if no entry in the list
-	    if [message] == "[]" {
-	       drop { }
-	    }
-	    
-	    json {
-	         source => "[message]"
-	         target => "message"
-	    }
-#	    ruby {
-#	    	code => "event.get('message').each{|m| m.set('type',event.get('type')}"
-#	    }
-	    split {
-	          field => "message"
-	          add_field => {
-      			"type" => "%{type}"
-      			"topic" => "%{topic}"
-    		  }
-	    }
-	    
-	    json {
-	         source => "message"
-	    }
-	    
-	    mutate { remove_field => [ "message" ] }
-    }
-    
-    # express timestamps in milliseconds instead of microseconds
-    if [closedLoopAlarmStart] {
-        ruby {
-            code => "
-                     if event.get('closedLoopAlarmStart').to_s.to_i(10) > 9999999999999
-                       event.set('closedLoopAlarmStart', event.get('closedLoopAlarmStart').to_s.to_i(10) / 1000)
-                     else
-                       event.set('closedLoopAlarmStart', event.get('closedLoopAlarmStart').to_s.to_i(10))
-                     end
-                    "
-        }
-        date {
-            match => [ "closedLoopAlarmStart", UNIX_MS ]
-            target => "closedLoopAlarmStart"
-        }
-    }
-
-    if [closedLoopAlarmEnd] {
-        ruby {
-            code => "
-                    if event.get('closedLoopAlarmEnd').to_s.to_i(10) > 9999999999999  
-                      event.set('closedLoopAlarmEnd', event.get('closedLoopAlarmEnd').to_s.to_i(10) / 1000)
-                    else
-                      event.set('closedLoopAlarmEnd', event.get('closedLoopAlarmEnd').to_s.to_i(10))
-                    end
-                    "
-        }
-        date {
-            match => [ "closedLoopAlarmEnd", UNIX_MS ]
-            target => "closedLoopAlarmEnd"
-        }
-
-    }
-    #"yyyy-MM-dd HH:mm:ss"
-    if [notificationTime] {
-       mutate {
-              gsub => [
-                   "notificationTime", " ", "T"
-              ]
-       }
-       date {
-            match => [ "notificationTime", ISO8601 ]
-            target => "notificationTime"
-       }
-    }
-}
-output {
-    stdout {
-        codec => rubydebug
-    }
-
-    if [http_request_failure] {
-        elasticsearch {
-            codec => "json"
-            hosts => ["${elasticsearch_hosts}"]
-            index => "errors-%{+YYYY.MM.DD}"
-            doc_as_upsert => true
-        }
-    } else {
-        elasticsearch {
-            codec => "json"
-            hosts => ["${elasticsearch_hosts}"]
-            index => "events-%{+YYYY.MM.DD}" # creates daily indexes
-            doc_as_upsert => true
-
-        }
-    }
-
-}