# How to filter based on log\_stream

**URL:** <https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780>\
**Category:** Logstash\
**Created:** [August 3, 2017, 9:08pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780 "2017-08-03T21:08:05Z")\
**Posts on this page:** 16\
**Page:** 1

<div class="post-metadata">

**Author:** ![bhedayatian](https://avatars.discourse-cdn.com/v4/letter/b/57b2e6/32.png) [@bhedayatian](https://discuss.elastic.co/u/bhedayatian)\
**Post date:** [August 3, 2017, 9:08pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/1 "2017-08-03T21:08:06Z")

</div>

hello. in aws cloud watch, i have group 1 that has 4 streams, how can i get logs from just one of the streams in logstash? i am using cloudwatch\_logs plugin in logstash.  
i have this

cloudwatch\_logs {  
log\_group =\> ["Group 1"]   
region =\> "us-west-2"  
access\_key\_id =\> "sfsdfsdf"  
secret\_access\_key =\> "sdsdfdsfsd"  
}  
im

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [August 4, 2017, 6:06am UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/2 "2017-08-04T06:06:03Z")

</div>

I'm not familiar with the cloudwatch\_logs input, but it looks like the stream name is being stored in the `[cloudwatch_logs][log_stream]` field so you can look at that field to e.g. drop all events except those from a particular stream (see [https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html#conditionals](https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html#conditionals)).

---

<div class="post-metadata">

**Author:** ![bhedayatian](https://avatars.discourse-cdn.com/v4/letter/b/57b2e6/32.png) [@bhedayatian](https://discuss.elastic.co/u/bhedayatian)\
**Post date:** [August 4, 2017, 7:57am UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/3 "2017-08-04T07:57:40Z")

</div>

Thanks Magnus, c  
i see ongoing streams in this format in logstash console

{  
"@timestamp" =\> 2017-05-26T22:44:12.951Z,  
"@version" =\> "1",  
"message" =\> "15:44:12.950 [main] INFO org.springframework.data.rest.webmvc.RepositoryRestHandlerMapping - Mapped "{[/{repository}/{id}],methods=[OPTIONS],produces=[application/hal+json || application/json || application/\*+json;charset=UTF-8]}" onto public org.springframework.http.ResponseEntity\<?\> org.springframework.data.rest.webmvc.RepositoryEntityController.optionsForItemResource(org.springframework.data.rest.webmvc.RootResourceInformation)",  
"cloudwatch\_logs" =\> {  
"event\_id" =\> "33358316657573528386365774610515871725880801939687604257",  
"log\_group" =\> "my Log-Group",  
"ingestion\_time" =\> 2017-05-26T22:44:15.081Z,  
"log\_stream" =\> "dev-container"  
}

how do i filter out just those have log\_stream=='my favorite\_stream'???

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [August 4, 2017, 8:01am UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/4 "2017-08-04T08:01:09Z")

</div>

> how do i filter out just those have log\_stream==‘my favorite\_stream’???

Use a drop filter and wrap it in a conditional as described in the documentation I linked to so events from your favorite stream aren't dropped.

---

<div class="post-metadata">

**Author:** ![bhedayatian](https://avatars.discourse-cdn.com/v4/letter/b/57b2e6/32.png) [@bhedayatian](https://discuss.elastic.co/u/bhedayatian)\
**Post date:** [August 4, 2017, 8:26am UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/5 "2017-08-04T08:26:20Z")

</div>

thanks. how about filtering for a period of time?  
is there something like  
if [ingestion\_time] \> '2017-01-01'  
??

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [August 4, 2017, 8:32am UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/6 "2017-08-04T08:32:43Z")

</div>

Pretty much, but for that to work I think you'll have to convert the `ingestion_time` field to a string (it's currently a timestamp). Use a mutate filter's `convert` option or that.

---

<div class="post-metadata">

**Author:** ![bhedayatian](https://avatars.discourse-cdn.com/v4/letter/b/57b2e6/32.png) [@bhedayatian](https://discuss.elastic.co/u/bhedayatian)\
**Post date:** [August 9, 2017, 5:31pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/7 "2017-08-09T17:31:57Z")

</div>

Hi Magnus, I am too close can you please help me here as i am not familiar with the syntax

```
 filter {   
  mutate {
       convert => ["ingestion_time", "String"]
 }
if [cloudwatch_logs][ingestion_time] < "2017-06-12" {
           drop{}
}

```

}

i cant get it to work

---

<div class="post-metadata">

**Author:** ![bhedayatian](https://avatars.discourse-cdn.com/v4/letter/b/57b2e6/32.png) [@bhedayatian](https://discuss.elastic.co/u/bhedayatian)\
**Post date:** [August 9, 2017, 5:37pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/8 "2017-08-09T17:37:05Z")

</div>

Error registering plugin {:plugin=\>"#\<LogStash::FilterDelegator:0xefe471f @id="3442afcb1ec1a741b185766e812106260160fc3c-2", @klass=LogStash::Filters::Mutate, @metric\_events=#\<

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [August 9, 2017, 6:39pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/9 "2017-08-09T18:39:12Z")

</div>

> ```
> convert => ["ingestion_time", "String"]
> 
> ```

Your field is named `[cloudwatch_logs][ingestion_time]`, not plain `ingestion_time`.

---

<div class="post-metadata">

**Author:** ![bhedayatian](https://avatars.discourse-cdn.com/v4/letter/b/57b2e6/32.png) [@bhedayatian](https://discuss.elastic.co/u/bhedayatian)\
**Post date:** [August 9, 2017, 8:36pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/10 "2017-08-09T20:36:35Z")

</div>

i tried convert =\> ["[cloudwatch\_logs][ingestion\_time]", "String"] and still get errors

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [August 10, 2017, 5:56am UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/11 "2017-08-10T05:56:23Z")

</div>

Post those errors in full (what you posted above is incomplete) and **make sure you post the logs as preformatted text so they doesn't get mangled**.

---

<div class="post-metadata">

**Author:** ![bhedayatian](https://avatars.discourse-cdn.com/v4/letter/b/57b2e6/32.png) [@bhedayatian](https://discuss.elastic.co/u/bhedayatian)\
**Post date:** [August 10, 2017, 3:08pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/12 "2017-08-10T15:08:29Z")

</div>

```
filter {   
 mutate {
   		convert => ["[cloudwatch_logs][ingestion_time]", "String"]
}  
    if [cloudwatch_logs][ingestion_time] < "2017-06-12" {
         drop{}
   }

```

}

and this is the error i get

[2017-08-10T08:07:09,286][ERROR][logstash.agent] Cannot create pipeline {:reason=\>"Expected one of #, =\> at line 17, column 12 (byte 397) after filter { \n \t mutate {\n \t\tconvert =\> ["[cloudwatch\_logs][ingestion\_time]", "String"]\n \t \n\n \n if "}

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [August 10, 2017, 3:12pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/13 "2017-08-10T15:12:29Z")

</div>

What you've posted here look okay but the problem could be elsewhere in your file.

---

<div class="post-metadata">

**Author:** ![bhedayatian](https://avatars.discourse-cdn.com/v4/letter/b/57b2e6/32.png) [@bhedayatian](https://discuss.elastic.co/u/bhedayatian)\
**Post date:** [August 10, 2017, 3:15pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/14 "2017-08-10T15:15:50Z")

</div>

Sending Logstash's logs to C:/elk-5.5.1/logstash-5.5.1/logs which is now configured via log4j2.properties  
[2017-08-10T08:15:07,831][INFO][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=\>{:removed=\>[], :added=\>[[http://localhost:9200/](http://localhost:9200/)]}}  
[2017-08-10T08:15:07,831][INFO][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck\_url=\>[http://localhost:9200/](http://localhost:9200/), :path=\>"/"}  
[2017-08-10T08:15:07,956][WARN][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=\>#Java::JavaNet::URI:0x1c2aa0cd}  
[2017-08-10T08:15:07,956][INFO][logstash.outputs.elasticsearch] Using mapping template from {:path=\>nil}  
[2017-08-10T08:15:08,019][INFO][logstash.outputs.elasticsearch] Attempting to install template {:manage\_template=\>{"template"=\>"logstash-_", "version"=\>50001, "settings"=\>{"index.refresh\_interval"=\>"5s"}, "mappings"=\>{"default"=\>{"\_all"=\>{"enabled"=\>true, "norms"=\>false}, "dynamic\_templates"=\>[{"message\_field"=\>{"path\_match"=\>"message", "match\_mapping\_type"=\>"string", "mapping"=\>{"type"=\>"text", "norms"=\>false}}}, {"string\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"string", "mapping"=\>{"type"=\>"text", "norms"=\>false, "fields"=\>{"keyword"=\>{"type"=\>"keyword", "ignore\_above"=\>256}}}}}], "properties"=\>{"@timestamp"=\>{"type"=\>"date", "include\_in\_all"=\>false}, "@version"=\>{"type"=\>"keyword", "include\_in\_all"=\>false}, "geoip"=\>{"dynamic"=\>true, "properties"=\>{"ip"=\>{"type"=\>"ip"}, "location"=\>{"type"=\>"geo\_point"}, "latitude"=\>{"type"=\>"half\_float"}, "longitude"=\>{"type"=\>"half\_float"}}}}}}}}  
[2017-08-10T08:15:08,035][INFO][logstash.outputs.elasticsearch] New Elasticsearch output {:class=\>"LogStash::Outputs::ElasticSearch", :hosts=\>[#Java::JavaNet::URI:0x63f87640]}  
[2017-08-10T08:15:08,035][ERROR][logstash.pipeline] Error registering plugin {:plugin=\>"#\<LogStash::FilterDelegator:0x4e700119 @id="a8f5b95a3626715416998f32d48250291ab92d15-2", @klass=LogStash::Filters::Mutate, @metric\_events=#\<LogStash::Instrument::NamespacedMetric:0x25947f1b @metric=#\<LogStash::Instrument::Metric:0x7c42abff @collector=#\<LogStash::Instrument::Collector:0x4360f9db @agent=nil, @metric\_store=#\<LogStash::Instrument::MetricStore:0xd564067 @store=#\<Concurrent:🗺0x0000000006407c entries=2 default\_proc=nil\>, @structured\_lookup\_mutex=#Mutex:0xbbc8a71, @fast\_lookup=#\<Concurrent:🗺0x00000000064080 entries=52 default\_proc=nil\>\>\>\>, @namespace\_name=[:stats, :pipelines, :main, :plugins, :filters, :"a8f5b95a3626715416998f32d48250291ab92d15-2", :events]\>, @logger=#\<LogStash::Logging::Logger:0x26224609 @logger=#Java::OrgApacheLoggingLog4jCore::Logger:0x76a27519\>, @filter=\<LogStash::Filters::Mutate convert=\>{"[cloudwatch\_logs][ingestion\_time]"=\>"String"}, id=\>"a8f5b95a3626715416998f32d48250291ab92d15-2", enable\_metric=\>true, periodic\_flush=\>false\>\>", :error=\>"translation missing: en.logstash.agent.configuration.invalid\_plugin\_register"}  
[2017-08-10T08:15:08,050][ERROR][logstash.agent] Pipeline aborted due to error {:exception=\>#\<LogStash::ConfigurationError: translation missing: en.logstash.agent.configuration.invalid\_plugin\_register\>, :backtrace=\>["C:/elk-5.5.1/logstash-5.5.1/vendor/bundle/jruby/1.9/gems/logstash-filter-mutate-3.1.5/lib/logstash/filters/mutate.rb:189:in `register'", "org/jruby/RubyHash.java:1342:in`each'", "C:/elk-5.5.1/logstash-5.5.1/vendor/bundle/jruby/1.9/gems/logstash-filter-mutate-3.1.5/lib/logstash/filters/mutate.rb:183:in `register'", "C:/elk-5.5.1/logstash-5.5.1/logstash-core/lib/logstash/pipeline.rb:281:in`register\_plugin'", "C:/elk-5.5.1/logstash-5.5.1/logstash-core/lib/logstash/pipeline.rb:292:in `register_plugins'", "org/jruby/RubyArray.java:1613:in`each'", "C:/elk-5.5.1/logstash-5.5.1/logstash-core/lib/logstash/pipeline.rb:292:in `register_plugins'", "C:/elk-5.5.1/logstash-5.5.1/logstash-core/lib/logstash/pipeline.rb:302:in`start\_workers'", "C:/elk-5.5.1/logstash-5.5.1/logstash-core/lib/logstash/pipeline.rb:226:in `run'", "C:/elk-5.5.1/logstash-5.5.1/logstash-core/lib/logstash/agent.rb:398:in`start\_pipeline'"]}  
[2017-08-10T08:15:08,175][INFO][logstash.agent] Successfully started Logstash API endpoint {:port=\>9600}  
[2017-08-10T08:15:11,066][WARN][logstash.agent] stopping pipeline {:id=\>"main"}

---

<div class="post-metadata">

**Author:** ![bhedayatian](https://avatars.discourse-cdn.com/v4/letter/b/57b2e6/32.png) [@bhedayatian](https://discuss.elastic.co/u/bhedayatian)\
**Post date:** [August 10, 2017, 3:17pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/15 "2017-08-10T15:17:41Z")

</div>

sorry about that. I missed a closing brace. now i see the real error. thanks for your help

---

<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:** [September 7, 2017, 3:17pm UTC](https://discuss.elastic.co/t/how-to-filter-based-on-log-stream/95780/16 "2017-09-07T15:17:43Z")

</div>

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