Process decoded data in more readable format

Elastic Stack Logstash
1

the decoded data from data streaming is coming into logstash.

"decoded": [
    "{\"headers\":{\"eventType\":\"com.data.globalIdentity.verb\",\"operation\":\"UPDATE\",\"messageType\":\"IDENTITY\",\"eventTime\":\"2025-07-15T15:25:01.814539994Z\",\"tenancyId\":\"idNumber\",\"serviceInstanceId\":\"instanceID\",\"opcRequestId\":\"opcNumber-e00ec\",\"eventTypeVersion\":\"1.0\",\"eventId\":\"xxbnms1-abc-111-xxab-hwy\"}
	
	]

used filter to make it in human readable format

filter {

ruby { code => 'event.set("decoded", Base64.decode64(event.get("message")))' }
mutate {
    gsub => [
        "decoded", "\"", "",
         "decoded", "\,", "
" # want to introduce new line here instead of carriage return but does Not work out
		 
		 ]
}

}

Current output in logstash console

    "message" => "decoded: [\n    {headers:{eventType:com.data.globalIdentity.verb\r\noperation:UPDATE\r\nmessageType:IDENTITY\r\neventTime:2025-07-15T15:25:01.814539994Z\r\ntenancyId:idNumber\r\nserviceInstanceId:instanceID\r\nopcRequestId:opcNumber-e00ec\r\neventTypeVersion:1.0\r\neventId:xxbnms1-abc-111-xxab-hwy}\n\t\n\t]",
         "event" => {
        "original" => "\"decoded\": [\n    \"{\"headers\":{\"eventType\":\"com.data.globalIdentity.verb\",\"operation\":\"UPDATE\",\"messageType\":\"IDENTITY\",\"eventTime\":\"2025-07-15T15:25:01.814539994Z\",\"tenancyId\":\"idNumber\",\"serviceInstanceId\":\"instanceID\",\"opcRequestId\":\"opcNumber-e00ec\",\"eventTypeVersion\":\"1.0\",\"eventId\":\"xxbnms1-abc-111-xxab-hwy\"}\n\t\n\t]"
    }

Desired Output

       eventType:com.data.globalIdentity.verb
	   operation:UPDATE
	   messageType:IDENTITY
	   eventTime:2025-07-15T15:25:01.814539994Z
	   tenancyId:idNumber
	   serviceInstanceId:instanceID
	   opcRequestId:opcNumber-e00ec
	   eventTypeVersion:1.0
	   eventId:xxbnms1-abc-111-xxab-hwy
2

Use a kv filter. For example

    mutate {
        gsub => [
            "message", "decoded: \[", "",
            "message", "{headers:{", "",
            "message", "}", ""

        ]
    }
    kv {
        field_split => "\n"
        value_split => ":"
        trim_key => "\s"
    }
3

Current input.conf (used the filter as provided in previous kv filter example)

input {
  kafka {
    bootstrap_servers => "abcd_cloud.com:9092"
    topics => "_Streaming"
    security_protocol => "SASL_SSL"
    sasl_mechanism => "PLAIN"
    jaas_path => "<Path>\logstash\logstash-8.17.2\jaas.conf"
    auto_offset_reset => "earliest"
    consumer_threads => 1
  }
}
filter {

ruby { code => 'event.set("decoded", Base64.decode64(event.get("message")))' }
mutate {
        gsub => [
            "message", "decoded: \[", "",
            "message", "{headers:{", "",
            "message", "}", ""

        ]
    }
    kv {
        field_split => "\n"
        value_split => ":"
        trim_key => "\s"
    }
}
output {
  stdout {
    codec => rubydebug
  }
  elasticsearch {
    hosts => ["http://localhost:<port>"]
  }
}

output in logstash (truncated for brevity, privacy)

       "decoded" => "{\"headers\":{\"eventType\":\"com.ity.changed\",\"operation\":\"UPDATE\",\"messageType\":\"IDENTITY\",\"eventTime\":\"2025-07-15T15:25:01.814539994Z\",\"tenancyId\":\"ocid1.tehlpa\",\"serviceInstanceId\":\"omy4q\",\"opcRequestId\":\"e0f-7024\",\"eventTypeVersion\":\"1.0\",\"eventId\":\"c3-1c67c\"},\"data\":\"{\\\"globalIdentity\\\":{\\\"identity\\\":{\\\"addresses\\\":[{}],\\\"agRisk\\\":{\\\"value\\\":0},\\\"agStatus\\\":\\\"AG_ACTIVE\\\",\\\"agSubType\\\":\\\"CONSUMER\\\",\\\"agOrganizations\\\":[{\\\"displayName\\\":\\\"allActiveUsers\\\"}],\\\"customAttributes\\\":{\\\"isCorrelated\\\":true,\\\"domainName\\\":\\\"CloudService\\\",\\\"id\\\":\\\"oa\\\",\\\"lastSuccessfulLoginTime\\\":0.0,\\\"compartmentID\\\":\\\"olpa\\\",\\\"compartmentName\\\":\\\"sas\\\",\\\"cloudAccountName\\\":\\\"sas\\\"},\\\"department\\\":\\\"UIdept001\\\",\\\"displayName\\\":\\\"iSL type016\\\",\\\"domainOCID\\\":\\\"o2a\\\",\\\"emails\\\":[{\\\"value\\\":\\\"ktype016@ktype016.com\\\"}],\\\"name\\\":{\\\"familyName\\\":\\\"type016\\\",\\\"givenName\\\":\\\"f_iSL\\\"},\\\"phoneNumbers\\\":[],\\\"primaryEmail\\\":\\\"ktype016@ktype016.com\\\",
4

The input string (base64 encoded), this is decoded and then set in decoded field.

eyJoZWFkZXJzIjp7ImV2ZW50VHlwZSI6ImNvbS5vcmFjbGUuaWRtLmFnY3MuZGF0YS5lbmFibGVtZW50Lmdsb2JhbElkZW50aXR5LmNoYW5nZWQiLCJvcGVyYXRpb24iOiJVUERBVEUiLCJtZXNzYWdlVHlwZSI6IklERU5USVRZIiwiZXZlbnRUaW1lIjoiMjAyNS0wNy0xNVQxNToyNTowMS44MTQ1Mzk5OTRaIiwid

modified input.conf as the gstub was not working as desired.
Didn't substitute comma and used it in the field_split

filter {

ruby { code => 'event.set("decoded", Base64.decode64(event.get("message")))' }
mutate {
        gsub => [
		
		"decoded", "\"", "",
		 "decoded", "{headers:{", "",
		 "decoded", "}", ""
		
        ]
		
 }
		kv {
        field_split => ","
		value_split => ":"
		trim_key => "\s"
		source => "decoded"

    }
		
		
    }

output {
  stdout {
    codec => rubydebug
  }
  elasticsearch {
    hosts => ["http://localhost:9200"]
  }
}

Though not together but all the fields are listed in logStash console.

{
     "@timestamp" => 2025-07-17T17:13:20.947789600Z,
       "@version" => "1",
        "decoded" => "eventType:com.oracle.idm.agcs.data.enablement.globalIdentity.changed,operation:UPDATE,messageType:IDENTITY,eventTime:2025-07-15T15:25:01.814539994Z,",
      "**operation**" => "UPDATE",
      "**eventType**" => "com.oracle.idm.agcs.data.enablement.globalIdentity.changed",
        "message" => "eyJoZWFkZXJzIjp7ImV2ZW50VHlwZSI6ImNvbS5vcmFjbGUuaWRtLmFnY3MuZGF0YS5lbmFibGVtZW50Lmdsb2JhbElkZW50aXR5LmNoYW5nZWQiLCJvcGVyYXRpb24iOiJVUERBVEUiLCJtZXNzYWdlVHlwZSI6IklERU5USVRZIiwiZXZlbnRUaW1lIjoiMjAyNS0wNy0xNVQxNToyNTowMS44MTQ1Mzk5OTRaIiwid",
          "event" => {
        "original" => "eyJoZWFkZXJzIjp7ImV2ZW50VHlwZSI6ImNvbS5vcmFjbGUuaWRtLmFnY3MuZGF0YS5lbmFibGVtZW50Lmdsb2JhbElkZW50aXR5LmNoYW5nZWQiLCJvcGVyYXRpb24iOiJVUERBVEUiLCJtZXNzYWdlVHlwZSI6IklERU5USVRZIiwiZXZlbnRUaW1lIjoiMjAyNS0wNy0xNVQxNToyNTowMS44MTQ1Mzk5OTRaIiwid"
    },
    "**messageType**" => "IDENTITY",
      "**eventTime**" => "2025-07-15T15:25:01.814539994Z"
}
5
  1. Is there any way to print all the fields at one time instead of being spread across console output ?
  2. Another issue is an error reported for nested key:value pairs {} due to which multiple fields aren't printed in logs though are reflecting under the decoded field
    example [present in the decoded field]
data:{\\globalIdentity\\:{\\identity\\:{\\addresses\\:[{}],\\agRisk\\:{\\value\\:0},\\agStatus\\:\\AG_ACTIVE\\,\\agSubType\\:\\CONSUMER\\,\\agOrganizations\\:[{\\displayName\\:\\allActiveUsers\\}],\\customAttributes\\:{\\isCorrelated\\:true,\\domainName\\:\\CloudService\\,\\id\\:\\id1.user.o1..22a\\,\\lastSuccessfulLoginTime\\:0.0,\\compartmentID\\:\\id1.tenancy.o1..lpa\\,\\compartmentName\\:\\sas\\,\\cloudAccountName\\:\\sas\\},\\department\\:\\UIdept001\\,\\displayName\\:\\KGiSL type016\\

Error

Exception while parsing KV {:exception=>"Invalid FieldReference: `\\agStatus\\:\\AG_ACTIVE\\,\\agSubType\\:\\CONSUMER\\,\\agOrganizations\\:[{\\displayName\\`"}
6

For the example you gave you could try

input { generator { count => 1 lines => [ 'eyJoZWFkZXJzIjp7ImV2ZW50VHlwZSI6ImNvbS5vcmFjbGUuaWRtLmFnY3MuZGF0YS5lbmFibGVtZW50Lmdsb2JhbElkZW50aXR5LmNoYW5nZWQiLCJvcGVyYXRpb24iOiJVUERBVEUiLCJtZXNzYWdlVHlwZSI6IklERU5USVRZIiwiZXZlbnRUaW1lIjoiMjAyNS0wNy0xNVQxNToyNTowMS44MTQ1Mzk5OTRaIiwid' ] } }

output { stdout { codec => rubydebug { metadata => false } } }
filter {
    ruby { code => 'event.set("[@metadata][decoded]", Base64.decode64(event.get("message")))' }
    mutate { gsub => [ "[@metadata][decoded]", '{"headers":{', "" ] }
    kv {
        source => "[@metadata][decoded]"
        field_split => ","
        value_split => ":"
        trim_key => '\s"'
    }
}

It's hard to say much about the nested key:value pairs since you only show the output after it is processed. What does the encoded message look like for that one?

7

The streaming services sends a base64 encoded data as below

eyJoZWFkZXJzIjp7ImV2ZW50VHlwZSI6ImNvbS5kYXRhLmVuYWJsZW1lbnQuZ2xvYmFsSWRlbnRpdHkuY2hhbmdlZCIsIm9wZXJhdGlvbiI6IlVQREFURSIsIm1lc3NhZ2VUeXBlIjoiSURFTlRJVFkiLCJldmVudFRpbWUiOiIyMDI1LTA3LTE1VDE1OjI1OjAxLjgxNDUzOTk5NFoiLCJ0ZW5hbmN5SWQiOiJpZDEudGVuYW5jeS5vYzEuLmhscGEiLCJzZXJ2aWNlSW5zdGFuY2VJZCI6ImlkMS5pbnN0YW5jZS55NHEiLCJvcGNSZXF1ZXN0SWQiOiJlYy0xMmUtNDEtYWYtNzI0IiwiZXZlbnRUeXBlVmVyc2lvbiI6IjEuMCIsImV2ZW50SWQiOiJjMzEtYmMtNDktYWYtZjJjIn0sImRhdGEiOiJ7XCJnbG9iYWxJZGVudGl0eVwiOntcImlkZW50aXR5XCI6e1wiYWRkcmVzc2VzXCI6W3t9XSxcImFnUmlza1wiOntcInZhbHVlXCI6MH0sXCJhZ1N0YXR1c1wiOlwiQUdfQUNUSVZFXCIsXCJhZ1N1YlR5cGVcIjpcIkNPTlNVTUVSXCIsXCJhZ09yZ2FuaXphdGlvbnNcIjpbe1wiZGlzcGxheU5hbWVcIjpcImFsbEFjdGl2ZVVzZXJzXCJ9XSxcImN1c3RvbUF0dHJpYnV0ZXNcIjp7XCJpc0NvcnJlbGF0ZWRcIjp0cnVlLFwiZG9tYWluTmFtZVwiOlwiQ2xvdWRTZXJ2aWNlXCIsXCJvY2lkXCI6XCJpZDEudXIubzEuLjIyYVwiLFwibGFzdFN1Y2Nlc3NmdWxMb2dpblRpbWVcIjowLjAsXCJjb21wYXJ0bWVudE9DSURcIjpcImlkMS50ZW5hbmN5Lm8xLi5scGFcIixcImNvbXBhcnRtZW50TmFtZVwiOlwic2FzXCIsXCJjbG91ZEFjY291bnROYW1lXCI6XCJzYXNcIn0sXCJkZXBhcnRtZW50XCI6XCJVSWRlcHQwMDFcIixcImRpc3BsYXlOYW1lXCI6XCJLR2lTTCB0eXBlMDE2XCIsXCJkb21haW5PQ0lEXCI6XCJpZDEuZG9tYWluLm8xLi5uMmFcIixcImVtYWlsc1wiOlt7XCJ2YWx1ZVwiOlwia3R5cGUwMTZAa3R5cGUwMTYuY29tXCJ9XSxcIm5hbWVcIjp7XCJmYW1pbHlOYW1lXCI6XCJ0eXBlMDE2XCIsXCJnaXZlbk5hbWVcIjpcImZfS0dpU0xcIn0sXCJwaG9uZU51bWJlcnNcIjpbXSxcInByaW1hcnlFbWFpbFwiOlwia3R5cGUwMTZAa3R5cGUwMTYuY29tXCIsXCJyZWdpb25cIjpcInB4OnVzLW9pLTFcIixcInN0YXR1c1wiOlwiQWN0aXZlXCIsXCJ1c2VyTmFtZVwiOlwia3R5cGUwMTZcIixcInVzZXJUeXBlXCI6XCJlQ2hhbmdlMDAxX3R5cGUwMTZcIn0sXCJpZFwiOlwiZ2xvYmFsSWQuSUYuNzRjMC1lNC00Yi04NS01Y2U3LmM1NjVkXCIsXCJ0YXJnZXRJZGVudGl0aWVzXCI6W3tcInRhcmdldElkXCI6XCI0ZjMtYzUtNDktYjctODBkOFwiLFwiaWRlbnRpdHlcIjp7XCJhZGRyZXNzZXNcIjpbe1wibG9jYWxpdHlcIjpcImNpdHlfVUkwMDFcIixcInR5cGVcIjpcIndvcmtcIn1dLFwiYWdPcmdhbml6YXRpb25zXCI6W10sXCJjdXN0b21BdHRyaWJ1dGVzXCI6e1wiY29tcGFydG1lbnRGUU5cIjpcInNlaHViamFwYWNpYWFzXCIsXCJvY2lkXCI6XCJpZDEudXNlci5vYzEuLjIyYVwiLFwibGFzdFN1Y2Nlc3NmdWxMb2dpblRpbWVcIjowLjAsXCJjbG91ZEFjY291bnROYW1lXCI6XCJzYXNcIixcImdyb3VwQXNzaWdubWVudHNcIjpbXSxcInJvbGVBc3NpZ25tZW50c1wiOlt7XCJjcmVhdGVEYXRlXCI6MS43NTA5Mzg5NDlFOSxcImdyYW50XCI6e1widmFsdWVcIjpcIjAxN2FjNWJjYjNcIn0sXCJwcm92aXNpb25pbmdNZWNoYW5pc21cIjpcIkFETUlOSVNUUkFUT1JfVE9fVVNFUlwiLFwicm9sZVwiOntcImRpc3BsYXlOYW1lXCI6XCJhcHBSb2xlc1wiLFwidmFsdWVcIjpcImIzNTZcIn0sXCJ1aWRcIjpcImIzNlwifV0sXCJjb21wYXJ0bWVudE9DSURcIjpcImlkMS50ZW5hbmN5Lm8xLi5obHBhXCIsXCJpZGNzSWRcIjpcImVmNDI5ZlwiLFwibGFzdE1vZGlmaWVkQnlcIjp7XCJkaXNwbGF5TmFtZVwiOlwiZGFzaCBlbGFzdGljXCIsXCJyZWZcIjpcImh0dHBzOi8vaXMtOTAudHkuZWNsb3VkLmNvbTo0NDMvYWRtaW4vdjEvVXNlcnMvMDNlXCIsXCJyZXNvdXJjZVR5cGVcIjpcIkFDQ09VTlRcIixcInZhbHVlXCI6XCJpZDEudXNlci5vYzEuLnJhXCJ9LFwiY29tcGFydG1lbnROYW1lXCI6XCJzYXNcIixcImRvbWFpbk5hbWVcIjpcInlDbG91ZFNlcnZpY2VcIixcInRhcmdldEFjY291bnRzXCI6W1wiZWY5NDI5ZlwiXSxcInByZXZpb3VzU3VjY2Vzc2Z1bExvZ2luVGltZVwiOjAuMCxcIm9wZXJhdGlvblR5cGVcIjpcIkNSRUFURV9PUl9VUERBVEVcIixcInRhZ3NcIjpbXSxcImNyZWF0ZWRCeVwiOntcImRpc3BsYXlOYW1lXCI6XCJmaXJzdCBhZ2NzLXVzZXJcIixcInJlZlwiOlwiaHR0cHM6Ly9pcy05NjAuZWNsb3VkLmNvbTo0NDMvYWRtaW4vdjEvVXNlcnMvUEkgVXNlclwiLFwicmVzb3VyY2VUeXBlXCI6XCJBQ0NPVU5UXCIsXCJ2YWx1ZVwiOlwiaWQxLnVzZXIub2MxLi5qYVwifSxcInByb3Zpc2lvbmVkRGF0ZVwiOjAuMH0sXCJkZXBhcnRtZW50XCI6XCJVSWRlcHQwMDFcIixcImRpc3BsYXlOYW1lXCI6XCJmX0tHaVNMIHR5cGUwMTZcIixcImRvbWFpbk9DSURcIjpcImlkMS5kb21haW4ub2MxLi5uMmFcIixcImVtYWlsc1wiOlt7XCJwcmltYXJ5XCI6ZmFsc2UsXCJzZWNvbmRhcnlcIjpmYWxzZSxcInR5cGVcIjpcInJlY292ZXJ5XCIsXCJ2YWx1ZVwiOlwia3R5cGUwMTZAa3R5cGUwMTYuY29tXCIsXCJ2ZXJpZmllZFwiOmZhbHNlfSx7XCJwcmltYXJ5XCI6dHJ1ZSxcInNlY29uZGFyeVwiOmZhbHNlLFwidHlwZVwiOlwid29ya1wiLFwidmFsdWVcIjpcImt0eXBlMDE2QGt0eXBlMDE2LmNvbVwiLFwidmVyaWZpZWRcIjpmYWxzZX1dLFwibmFtZVwiOntcImZvcm1hdHRlZFwiOlwia3R5cGUwMTZcIn0sXCJvcmdhbml6YXRpb25cIjp7fSxcInBob25lTnVtYmVyc1wiOltdLFwicHJpbWFyeUVtYWlsXCI6XCJrdHlwZTAxNkBrdHlwZTAxNi5jb21cIixcInJlZ2lvblwiOlwicGh4OnVzLXBob2VuaXgtMVwiLFwic3RhdHVzXCI6XCJEaXNhYmxlZFwiLFwidGl0bGVcIjpcIlVJX3RpdGxlMDAxXCIsXCJ1c2VyTmFtZVwiOlwia3R5cGUwMTZcIn0sXCJleHRlcm5hbElkXCI6XCJlZmViMWFmOTgxZTg0MDYxYWYyN2QyMTRjNmE5NDI5ZlwiLFwiaWRcIjpcInRhcmdldElkLmFjY291bnQuT0kuNGYzLWM1LTQ5LWI3LTgwY2Q4LjIwMDEyMVwiLFwiZG9tYWluSWRcIjpcInJlc291cmNlLkkuNDMtYzUtNDktYjctOGQ4LjFkOFwifV19fSJ9

which decodes (base64 decoded in notepad++) into below nested data

{"headers":{"eventType":"com.data.enablement.globalIdentity.changed","operation":"UPDATE","messageType":"IDENTITY","eventTime":"2025-07-15T15:25:01.814539994Z","tenancyId":"id1.tenancy.oc1..hlpa","serviceInstanceId":"id1.instance.y4q","opcRequestId":"ec-12e-41-af-724","eventTypeVersion":"1.0","eventId":"c31-bc-49-af-f2c"},"data":"{\"globalIdentity\":{\"identity\":{\"addresses\":[{}],\"agRisk\":{\"value\":0},\"agStatus\":\"AG_ACTIVE\",\"agSubType\":\"CONSUMER\",\"agOrganizations\":[{\"displayName\":\"allActiveUsers\"}],\"customAttributes\":{\"isCorrelated\":true,\"domainName\":\"CloudService\",\"ocid\":\"id1.ur.o1..22a\",\"lastSuccessfulLoginTime\":0.0,\"compartmentOCID\":\"id1.tenancy.o1..lpa\",\"compartmentName\":\"sas\",\"cloudAccountName\":\"sas\"},\"department\":\"UIdept001\",\"displayName\":\"KGiSL type016\",\"domainOCID\":\"id1.domain.o1..n2a\",\"emails\":[{\"value\":\"ktype016@ktype016.com\"}],\"name\":{\"familyName\":\"type016\",\"givenName\":\"f_KGiSL\"},\"phoneNumbers\":[],\"primaryEmail\":\"ktype016@ktype016.com\",\"region\":\"px:us-oi-1\",\"status\":\"Active\",\"userName\":\"ktype016\",\"userType\":\"eChange001_type016\"},\"id\":\"globalId.IF.74c0-e4-4b-85-5ce7.c565d\",\"targetIdentities\":[{\"targetId\":\"4f3-c5-49-b7-80d8\",\"identity\":{\"addresses\":[{\"locality\":\"city_UI001\",\"type\":\"work\"}],\"agOrganizations\":[],\"customAttributes\":{\"compartmentFQN\":\"sehubjapaciaas\",\"ocid\":\"id1.user.oc1..22a\",\"lastSuccessfulLoginTime\":0.0,\"cloudAccountName\":\"sas\",\"groupAssignments\":[],\"roleAssignments\":[{\"createDate\":1.750938949E9,\"grant\":{\"value\":\"017ac5bcb3\"},\"provisioningMechanism\":\"ADMINISTRATOR_TO_USER\",\"role\":{\"displayName\":\"appRoles\",\"value\":\"b356\"},\"uid\":\"b36\"}],\"compartmentOCID\":\"id1.tenancy.o1..hlpa\",\"idcsId\":\"ef429f\",\"lastModifiedBy\":{\"displayName\":\"dash elastic\",\"ref\":\"https://is-90.ty.ecloud.com:443/admin/v1/Users/03e\",\"resourceType\":\"ACCOUNT\",\"value\":\"id1.user.oc1..ra\"},\"compartmentName\":\"sas\",\"domainName\":\"yCloudService\",\"targetAccounts\":[\"ef9429f\"],\"previousSuccessfulLoginTime\":0.0,\"operationType\":\"CREATE_OR_UPDATE\",\"tags\":[],\"createdBy\":{\"displayName\":\"first agcs-user\",\"ref\":\"https://is-960.ecloud.com:443/admin/v1/Users/PI User\",\"resourceType\":\"ACCOUNT\",\"value\":\"id1.user.oc1..ja\"},\"provisionedDate\":0.0},\"department\":\"UIdept001\",\"displayName\":\"f_KGiSL type016\",\"domainOCID\":\"id1.domain.oc1..n2a\",\"emails\":[{\"primary\":false,\"secondary\":false,\"type\":\"recovery\",\"value\":\"ktype016@ktype016.com\",\"verified\":false},{\"primary\":true,\"secondary\":false,\"type\":\"work\",\"value\":\"ktype016@ktype016.com\",\"verified\":false}],\"name\":{\"formatted\":\"ktype016\"},\"organization\":{},\"phoneNumbers\":[],\"primaryEmail\":\"ktype016@ktype016.com\",\"region\":\"phx:us-phoenix-1\",\"status\":\"Disabled\",\"title\":\"UI_title001\",\"userName\":\"ktype016\"},\"externalId\":\"efeb1af981e84061af27d214c6a9429f\",\"id\":\"targetId.account.OI.4f3-c5-49-b7-80cd8.200121\",\"domainId\":\"resource.I.43-c5-49-b7-8d8.1d8\"}]}}"}

Currently, the filter in file : input.conf is able to partially extract the data out of the base64 decoded data

input {
  kafka {
    bootstrap_servers => "ecloud.com:9092"
    topics => "_Streaming"
    security_protocol => "SASL_SSL"
    sasl_mechanism => "PLAIN"
    jaas_path => "<PATH>\logstash\logstash-8.17.2\jaas.conf"
    auto_offset_reset => "earliest"
    consumer_threads => 1
  }
}
filter {

ruby { code => 'event.set("decoded", Base64.decode64(event.get("message")))' }
mutate {
        gsub => [
		
		"decoded", "\"", "",
         #"decoded", "\,", "#",
		 "decoded", "{headers:{", ""
		 #"decoded", "}", ""
		
        ]
		
 }
		kv {
        field_split => ","
		value_split => ":"
		trim_key => "\s"
		source => "decoded"

    }
		
		
    }


output {
  stdout {
    codec => rubydebug
  }
  elasticsearch {
    hosts => ["http://localhost:9200"]
  }
}

logstash output of the above encoded data using highlighted filter

[2025-07-18T15:00:28,451][WARN ][logstash.filters.kv      ][main][c378ddfa19255977b80ebc28413aa3defed410d20c6ca4ef288fe3257e2ca0e7] Exception while parsing KV {:exception=>"Invalid FieldReference: `\\agStatus\\:\\AG_ACTIVE\\,\\agSubType\\:\\CONSUMER\\,\\agOrganizations\\:[{\\displayName\\`"}
{
                    "event" => {
        "original" => "<base64_encodedData>"
    },
             "opcRequestId" => "ec-12e-41-af-724",
                     "tags" => [
        [0] "_kv_filter_error"
    ],
                "operation" => "UPDATE",
                "eventType" => "com.data.enablement.globalIdentity.changed",
                  "decoded" => "eventType:com.data.enablement.globalIdentity.changed,operation:UPDATE,messageType:IDENTITY,eventTime:2025-07-15T15:25:01.814539994Z,tenancyId:id1.tenancy.oc1..hlpa,serviceInstanceId:id1.instance.y4q,opcRequestId:ec-12e-41-af-724,eventTypeVersion:1.0,eventId:c31-bc-49-af-f2c},data:{\\globalIdentity\\:{\\identity\\:{\\addresses\\:[{}],\\agRisk\\:{\\value\\:0},\\agStatus\\:\\AG_ACTIVE\\,\\agSubType\\:\\CONSUMER\\,\\agOrganizations\\:[{\\displayName\\:\\allActiveUsers\\}],\\customAttributes\\:{\\isCorrelated\\:true,\\domainName\\:\\CloudService\\,\\ocid\\:\\id1.ur.o1..22a\\,\\lastSuccessfulLoginTime\\:0.0,\\compartmentOCID\\:\\id1.tenancy.o1..lpa\\,\\compartmentName\\:\\sas\\,\\cloudAccountName\\:\\sas\\},\\department\\:\\UIdept001\\,\\displayName\\:\\KGiSL type016\\,\\domainOCID\\:\\id1.domain.o1..n2a\\,\\emails\\:[{\\value\\:\\ktype016@ktype016.com\\}],\\name\\:{\\familyName\\:\\type016\\,\\givenName\\:\\f_KGiSL\\},\\phoneNumbers\\:[],\\primaryEmail\\:\\ktype016@ktype016.com\\,\\region\\:\\px:us-oi-1\\,\\status\\:\\Active\\,\\userName\\:\\ktype016\\,\\userType\\:\\eChange001_type016\\},\\id\\:\\globalId.IF.74c0-e4-4b-85-5ce7.c565d\\,\\targetIdentities\\:[{\\targetId\\:\\4f3-c5-49-b7-80d8\\,\\identity\\:{\\addresses\\:[{\\locality\\:\\city_UI001\\,\\type\\:\\work\\}],\\agOrganizations\\:[],\\customAttributes\\:{\\compartmentFQN\\:\\sehubjapaciaas\\,\\ocid\\:\\id1.user.oc1..22a\\,\\lastSuccessfulLoginTime\\:0.0,\\cloudAccountName\\:\\sas\\,\\groupAssignments\\:[],\\roleAssignments\\:[{\\createDate\\:1.750938949E9,\\grant\\:{\\value\\:\\017ac5bcb3\\},\\provisioningMechanism\\:\\ADMINISTRATOR_TO_USER\\,\\role\\:{\\displayName\\:\\appRoles\\,\\value\\:\\b356\\},\\uid\\:\\b36\\}],\\compartmentOCID\\:\\id1.tenancy.o1..hlpa\\,\\idcsId\\:\\ef429f\\,\\lastModifiedBy\\:{\\displayName\\:\\dash elastic\\,\\ref\\:\\https://is-90.ty.ecloud.com:443/admin/v1/Users/03e\\,\\resourceType\\:\\ACCOUNT\\,\\value\\:\\id1.user.oc1..ra\\},\\compartmentName\\:\\sas\\,\\domainName\\:\\yCloudService\\,\\targetAccounts\\:[\\ef9429f\\],\\previousSuccessfulLoginTime\\:0.0,\\operationType\\:\\CREATE_OR_UPDATE\\,\\tags\\:[],\\createdBy\\:{\\displayName\\:\\first agcs-user\\,\\ref\\:\\https://is-960.ecloud.com:443/admin/v1/Users/PI User\\,\\resourceType\\:\\ACCOUNT\\,\\value\\:\\id1.user.oc1..ja\\},\\provisionedDate\\:0.0},\\department\\:\\UIdept001\\,\\displayName\\:\\f_KGiSL type016\\,\\domainOCID\\:\\id1.domain.oc1..n2a\\,\\emails\\:[{\\primary\\:false,\\secondary\\:false,\\type\\:\\recovery\\,\\value\\:\\ktype016@ktype016.com\\,\\verified\\:false},{\\primary\\:true,\\secondary\\:false,\\type\\:\\work\\,\\value\\:\\ktype016@ktype016.com\\,\\verified\\:false}],\\name\\:{\\formatted\\:\\ktype016\\},\\organization\\:{},\\phoneNumbers\\:[],\\primaryEmail\\:\\ktype016@ktype016.com\\,\\region\\:\\phx:us-phoenix-1\\,\\status\\:\\Disabled\\,\\title\\:\\UI_title001\\,\\userName\\:\\ktype016\\},\\externalId\\:\\efeb1af981e84061af27d214c6a9429f\\,\\id\\:\\targetId.account.OI.4f3-c5-49-b7-80cd8.200121\\,\\domainId\\:\\resource.I.43-c5-49-b7-8d8.1d8\\}]}}}",
                "tenancyId" => "id1.tenancy.oc1..hlpa",
               "@timestamp" => 2025-07-18T09:30:28.331015300Z,
         "eventTypeVersion" => "1.0",
                  "eventId" => "c31-bc-49-af-f2c}",
                 "@version" => "1",
    "\\agRisk\\:{\\value\\" => "0}",
              "messageType" => "IDENTITY",
                "eventTime" => "2025-07-15T15:25:01.814539994Z",
        "serviceInstanceId" => "id1.instance.y4q",
                     "data" => "{\\globalIdentity\\:{\\identity\\:{\\addresses\\:[{}]",
                  "message" => "<base64_encodedData>"
}

if the above data (decoded) is processed in a human readable format in the logstash itself, then it'll be easier to use analytics feature in kibana too.

8

For that particular value it would be much easier to do

    ruby { code => 'event.set("[@metadata][decoded]", Base64.decode64(event.get("message")))' }

    json { source => "[@metadata][decoded]" }
    json { source => "data" remove_field => [ "data" ]}

Possibly then use mutate and kv only if you get _jsonparsefailure tagged.