ELK (via API)

Introduction

FlowG allows you to view the logs your pipelines stored in streams. But the log viewer is pretty much barebones, as the main feature of FlowG is the pipeline itself. For that reason, you might want to forward logs to an ELK Stack, so that you can use Kibana to view your logs.

Fortunately, FlowG is able to forward logs to an ElasticSearch database.

In this tutorial, we will setup an ElasticSearch using Docker and a FlowG pipeline to forward logs to it.

Setting up ElasticSearch

First, run the Docker container:

docker run -d --name elasticsearch \
  -e "discovery.type=single-node" \
  -e "ELASTIC_PASSWORD=changeme" \
  -p 9200:9200 \
  docker.elastic.co/elasticsearch/elasticsearch:8.15.3

Once started, create a user:

curl \
    -k \
    -u elastic:changeme \
    -X POST https://localhost:9200/_security/role/index_manager \
    -H 'Content-Type: application/json' \
    -d '{"indices": [{"names": ["default"], "privileges": ["create_index","manage","read","write","view_index_metadata"]}]}'

curl \
  -k \
  -u elastic:changeme \
  -X POST https://localhost:9200/_security/user/flowg \
  -H 'Content-Type: application/json' \
  -d '{"password": "changeme", "roles": ["index_manager"]}'

Then, fetch the CA certificate:

docker exec elasticsearch cat ./config/certs/http_ca.crt

Setting up the FlowG pipeline

First, let's create an "Elastic Forwarder" named es, with the following configuration:

Forwarder TypeElastic Search

Indexdefault

Usernameflowg

Password••••••••••••

Addresshttps://localhost:9200

CA Certificate-----BEGIN CERTIFICATE----- ... -----END CERTIFICATE-----

Then, create a pipeline that forwards logs to the es forwarder:

SourceSYSLOG

Forwarderes

And that's it!

Testing

We can then run a simple query against ElasticSearch to check that our logs are being forwarded correctly:

logger -n localhost -P 5514 -t myapp 'hello world'
curl \
  -k \
  -u elastic:changeme \
  -X GET "https://localhost:9200/default/_search?pretty" \
  -H "Content-Type: application/json" \
  -d '{"query": {"match_all": {}}}'

We should receive the following response:

{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "default",
        "_id" : "5HqEo5gBIAeBpdMlfIgf",
        "_score" : 1.0,
        "_source" : {
          "timestamp" : "2025-08-13T13:00:25.296237816Z",
          "fields" : {
            "app_name" : "myapp",
            "client" : "127.0.0.1:58598",
            "facility" : "1",
            "hostname" : "***",
            "message" : "hello world",
            "msg_id" : "-",
            "priority" : "13",
            "proc_id" : "-",
            "severity" : "5",
            "structured_data" : "[timeQuality tzKnown=\"1\" isSynced=\"1\" syncAccuracy=\"866000\"]",
            "timestamp" : "2025-08-13 13:00:25.296014 +0000 UTC",
            "tls_peer" : "",
            "version" : "1"
          }
        }
      }
    ]
  }
}

Introduction​

Setting up ElasticSearch​

Setting up the FlowG pipeline​

Testing​

Introduction

Setting up ElasticSearch

Setting up the FlowG pipeline

Testing