Gist for elasticsearch and jq arcticle

example_without_logstash.md

First we index the numbers from 1 to 10 using json_lines and cryptocli in a temporary index.

seq 1 10 | \
bash json_lines.sh number 1 2>/dev/null | \
./cryptocli \
  -- stdin \
  -- line \
  -- elasticsearch-put \
    --index number_messages \
    --type number \
    --raw

Then, for each messages in the last 15 minutes in the temporary index, we deserialize the message field from json, set the right index to our final number index, set the right fields.type. Finally, we extract the number from the json_line and set it to the root of the source.

 cryptocli  \
  -- elasticsearch-get \
    --index number_messages \
    -- fork jq -rnc '
      inputs |
      ._source.json_lines = (._source.message | fromjson) |
      del(._source.message) |
      ._index = "number" |
      ._source.fields.type = ["json_lines", "number"] |
      select(._source.json_lines.status == "stopped") // ._source.number = (._source.json_lines.lines[0].line | tonumber) |
      del(._source.json_lines.lines[0].line)' \
    -- elasticsearch-put

logstash.conf

if "json_lines" in [fields][type] {
	json {
		source => "message"
		target => "[json_lines]"
	}

	if 
		[json_lines][type] and
		[json_lines][id] and
		[json_lines][status] {
			mutate {
				remove_field => ["[message]"]
			}

			if "%{[json_lines][type]}" not in [fields][type] {
				mutate {
					add_field => {
						"[fields][type]" => "%{[json_lines][type]}"
					}
				}
			}
	}
	if
		"number" in [fields][type]  and
		[json_lines][status] == "running" {
     			ruby {
				code => '
					event.set("number", event.get("[json_lines][lines][line][0]"))
					event.set("[json_lines][lines][line]" = [])
				'
			}
		}
	}
}

mapping.json

{
  "order": 0,
  "template": "json_lines-v1-*",
  "settings": {
    "index": {
      "number_of_shards": "1",
      "codec": "best_compression",
      "number_of_replicas": "0"
    }
  },
  "mappings": {
    "beats": {
      "dynamic": false,
      "dynamic_templates": [
        {
          "blackhole": {
            "mapping": {
              "index": false,
              "norms": false
            },
            "match": "*"
          }
        }
      ],
      "_all": {
        "enabled": false
      },
      "properties": {
        "@timestamp": {
          "type": "date"
        },
        "tags": {
          "type": "text",
          "norms": false,
          "fields": {
            "raw": {
              "type": "keyword"
            }
          }
        },
        "fields": {
          "properties": {
            "type": {
              "type": "keyword"
            }
          }
        },
        "beat": {
          "properties": {
            "name": {
              "type": "keyword"
            },
            "hostname": {
              "type": "keyword"
            }
          }
        },
        "json_lines": {
          "properties": {
            "username": {
              "type": "keyword"
            },
            "took": {
              "type": "float"
            },
            "created_at": {
              "type": "date"
            },
            "id": {
              "type": "keyword"
            },
            "status": {
              "type": "keyword"
            },
            "type": {
              "type": "keyword"
            },
            "length": {
              "type": "integer"
            },
            "lines": {
              "properties": {
                "line": {
                  "type": "text"
                },
                "line_offset": {
                  "type": "long"
                }
              }
            },
            "max_offset": {
              "type": "long"
            }
          }
        }
      }
    }
  },
  "aliases": {
    "json_lines": {},
    "json_lines_no_timestamp": {}
  }
}