# Logstash aggregate filter not working for GCP MySQL slow query logs

**URL:** <https://discuss.elastic.co/t/logstash-aggregate-filter-not-working-for-gcp-mysql-slow-query-logs/301103>\
**Category:** Logstash\
**Created:** [March 30, 2022, 2:17pm UTC](https://discuss.elastic.co/t/logstash-aggregate-filter-not-working-for-gcp-mysql-slow-query-logs/301103 "2022-03-30T14:17:18Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![Chris\_Pinto](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/chris_pinto/32/103721_2.png) [@Chris\_Pinto](https://discuss.elastic.co/u/Chris_Pinto)\
**Post date:** [March 30, 2022, 2:17pm UTC](https://discuss.elastic.co/t/logstash-aggregate-filter-not-working-for-gcp-mysql-slow-query-logs/301103/1 "2022-03-30T14:17:18Z")

</div>

Hi Team,

I'm trying to send Google Cloud MySQL slow query logs from GCP Cloud Logging to elastic through Logstash pubsub input.

The Slow query logs are in the JSON format and Pubsub has a limitation - it does not maintain order when the events are forwarded to the subscribers(Logstash).

Hence, I'm aggregating them based on a receiveTimestamp but no luck.

The source JSON events looks something like this.

```auto
{
	"@timestamp":"2022-03-29T04:00:24.177Z",
	"message": "original_event_3",
	"textPayload": "# Query_time: 10.183602 Lock_time: 0.000386 Rows_sent: 823492 Rows_examined: 825476",
	"tags": ["cloudsql-slow-log"],
	"receiveTimestamp": "2022-03-29T04:00:36.305655815Z",
	"cloud.project.id": "ABC"
}
{
	"@timestamp":"2022-03-29T04:00:30.177Z",
	"message": "original_event_1",
	"textPayload": "# Time: 2022-03-29T08:37:50.228345Z"
	"tags": ["cloudsql-slow-log"]
	"receiveTimestamp": "2022-03-29T04:00:36.305655815Z"
	"cloud.project.id": "ABC"
}
{
	"@timestamp":"2022-03-29T04:00:40.177Z",
	"message": "original_event_6",
	"textPayload": "FROM reporting.v_v3_profile_identities"
	"tags": ["cloudsql-slow-log"]
	"receiveTimestamp": "2022-03-29T04:00:36.305655815Z"
	"cloud.project.id": "ABC"
}
{
	"@timestamp":"2022-03-29T04:00:55.177Z",
	"message": "original_event_4",
	"textPayload": "SET timestamp=1648543070;"
	"tags": ["cloudsql-slow-log"]
	"receiveTimestamp": "2022-03-29T04:00:36.305655815Z"
	"cloud.project.id": "ABC"
}
.......

```

Desired output is

```auto

The output should look something like below:

{
	"@timestamp":"2022-03-29T04:00:30.177Z",
	"textPayload": "# Time: 2022-03-29T08:37:50.228345Z
					# User@Host: mule_transaction_report[mule_transaction_report] @ [194.255.15.66] thread_id: 666318 server_id: 1912109277
					# Query_time: 10.183602 Lock_time: 0.000386 Rows_sent: 823492 Rows_examined: 825476
					SET timestamp=1648543070;
					SELECT identifier as GCNUMBER
					FROM reporting.v_v3_profile_identities 
					where handle ='card_pos' 
					and identifier >= '80029914'
					and identifier <= '83539938'
					order by identifier asc;"
	"tags": ["cloudsql-slow-log"]
	"receiveTimestamp": "2022-03-29T04:00:36.305655815Z"
	"cloud.project.id": "ABC"
}

```

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [March 30, 2022, 2:24pm UTC](https://discuss.elastic.co/t/logstash-aggregate-filter-not-working-for-gcp-mysql-slow-query-logs/301103/2 "2022-03-30T14:24:01Z")

</div>

You need to share your logstash configuration with your aggregate filter.

But the aggregate filter will aggregate events based on a unique id in the order the filter receives the events, if your events are already out of order before entering the logstash pipeline, the aggregate filter won't fix that.

---

<div class="post-metadata">

**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\
**Post date:** [March 30, 2022, 6:53pm UTC](https://discuss.elastic.co/t/logstash-aggregate-filter-not-working-for-gcp-mysql-slow-query-logs/301103/3 "2022-03-30T18:53:53Z")

</div>

You could try

```
    aggregate {
        task_id => "%{receiveTimestamp}"
        push_map_as_event_on_timeout => true
        timeout_task_id_field => "receiveTimestamp"
        timeout => 3
        code => '
            map["textPayload"] ||= []
            map["@timestamp"] ||= event.get("@timestamp")
            map["tags"] ||= event.get("tags")
            map["cloud.project.id"] ||= event.get("cloud.project.id")
            i = event.get("message").sub(/\D+/, "").to_i
            map["textPayload"][i-1] = event.get("textPayload")
            event.cancel
        '
        timeout_code => '
            # This code operates on the generated event,
            # not on the map from which it is generated.
            event.set("textPayload", event.get("textPayload").join("\n"))
        '
    }

```

which for the 4 events you show (once they are fixed to be valid JSON) will result in

```
{
      "@timestamp" => 2022-03-29T04:00:24.177Z,
            "tags" => ["cloudsql-slow-log"],
"receiveTimestamp" => "2022-03-29T04:00:36.305655815Z",
"cloud.project.id" => "ABC",
     "textPayload" => "# Time: 2022-03-29T08:37:50.228345Z\n\n# Query_time: 10.183602 Lock_time: 0.000386 Rows_sent: 823492 Rows_examined: 825476\nSET timestamp=1648543070;\n\nFROM reporting.v_v3_profile_identities"
}
```

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [April 27, 2022, 6:54pm UTC](https://discuss.elastic.co/t/logstash-aggregate-filter-not-working-for-gcp-mysql-slow-query-logs/301103/4 "2022-04-27T18:54:23Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.