official way

configs/pilot_log.conf

@@ -8,20 +8,38 @@ filter {
 
   mutate {
     remove_field => [ "headers", "type", "program", "host" ]
-    gsub => ["message", "'", '"']
   }
 
-  json {
-    source => "message"
-    remove_field => [ "message" ]
-    remove_field => [ "port" ]
-    tag_on_failure => ["_jsonparsefailure"]
+
+  mutate {
+    rename => {"message" => "input_message"}
+  }
+  grok {
+        patterns_dir => "/usr/share/logstash/patterns/"
+        match => { "input_message" => "{%{GREEDYDATA:message_kv}, 'message': %{GREEDYDATA:message}}" }
+  }
+
+  kv {
+    source => "message_kv"
+    trim_key => "'"
+    trim_value => "',"
+    value_split => ":"
+  }
+  mutate {
+    convert => {"PilotTimeStamp" => "string"}
+  }
+  ruby {
+      code => "
+          timestamp_source = (event.get('[PilotTimeStamp]')).split(/\s|\./)
+          event.set('[@timestamp_nanoseconds]', ((Time.at(timestamp_source[0].to_i).getutc).strftime('%Y-%m-%dT%H:%M:%S.') +  timestamp_source[1] + 'Z') )
+          event.set('[nanoseconds_debug]', ((Time.at(timestamp_source[0].to_i).getutc).strftime('%Y-%m-%dT%H:%M:%S.') +  timestamp_source[1] + 'Z') )
+          event.set('[@timestamp]', LogStash::Timestamp.new(event.get('[@timestamp_nanoseconds]')) )
+        "
+    }
   }
-
-  # ruby { path => "/usr/share/logstash/filters/reformat.rb" }
 
   mutate {
-    remove_field => [ "version", "@version", "[extra][logger_name]","[extra][logstash_async_version]","[extra][process_name]","[extra][thread_name]"]
+    remove_field => [ "version", "@version", "[extra][logger_name]", "[extra][logstash_async_version]", "[extra][process_name]", "[extra][thread_name]"]
   }
 # extra.path has this random thing that should be shortened /scratch/atlas_MTFrHlIS/pilot3/pilot/util/realtimelogger.py
 }