Barito Flow

Component for handling flow of logs within a cluster. Support 2 modes:

• Producer, for receiving logs and forwarding it to Kafka
• Consumer, for consuming from kafka and forwarding it to Elasticsearch

Compatible with both gRPC and REST API. The use of REST API is optional and is implemented by using gRPC-gateway which work like a reverse-proxy server to translate RESTful HTTP API into gRPC. Barito flow infrastructure consists of producer and consumer.

Barito flow producer will turn on gRPC server and optionally REST gateway reverse proxy server. It will automatically create Kafka topic for the log if not exist yet. gRPC messages and services are declared in barito-proto repository.

Barito flow consumer will firstly create a topic event and generates the workers. Then based on the logs send, each topic inside this event topic will be created a cluster consumer separately. This cluster consumer will store the logs to Elasticsearch by calling a single store or bulk store Elasticsearch API. If the process has failed, Elasticsearch will halt all the workers and retry again after some backoff period. The halted workers only continue when the failed process success on the retry attempt.

Development Setup

If running on local machine with ARM (e.g. Apple M1, Apple M2) Chipset, run below command.

go env -w GOARCH=amd64

Fetch and build the project.

git clone https://github.com/BaritoLog/barito-flow
cd barito-flow
go build

Generate mock classes.

mockgen -source=flow/leaky_bucket.go -destination=mock/leaky_bucket.go -package=mock
mockgen -source=flow/kafka_admin.go -destination=mock/kafka_admin.go -package=mock
mockgen -source=flow/Vendor/github.com/sarama/sync_producer.go -destination=mock/sync_producer.go -package=mock

Running Test Stack using Docker Compose

First, you need to install Docker on your local machine. Then you can run docker-compose:

$ docker-compose -f docker/docker-compose.yml up -d

This will pull Elasticsearch, Kafka, and build producer and consumer image. The ports are mapped as if they are running on local machine.

Run Unit Tests

make test

Check Vulnerability

make vuln

Check Deadcode

make deadcode

Producer Mode

Responsible for:

• Receive logs by exposing an HTTP endpoint
• Produce message to kafka cluster

After the project is built, run:

./barito-flow producer

# or
./barito-flow p

Endpoints using REST gateway:

POST /produce

  {
    "context": {
        "kafka_topic": "kafka_topic",
        "kafka_partition": 1,
        "kafka_replication_factor": 1,
        "es_index_prefix": "es_index_prefix",
        "es_document_type": "es_document_type",
        "app_max_tps": 100,
        "app_secret": "app_secret"
    },
    "timestamp": "optional timestamp here",
    "content": {
        "hello": "world",
        "key": "value",
        "num": 100
    },
  }

POST /produce_batch

  {
    "context": {
        "kafka_topic": "kafka_topic",
        "kafka_partition": 1,
        "kafka_replication_factor": 1,
        "es_index_prefix": "es_index_prefix",
        "es_document_type": "es_document_type",
        "app_max_tps": 100,
        "app_secret": "app_secret"
    },
    "items": [
        {
            "content": {
                "timber_num": 1
            }
        },
        {
            "content": {
                "timber_num": 2
            }
        }
    ]
  }

These environment variables can be modified to customize its behaviour.

Consumer Mode

Responsible for:

• Consume logs from kafka
• Commit logs to elasticsearch

After the project is built, run:

./barito-flow Consumer

# or
./barito-flow c

These environment variables can be modified to customize its behaviour.

NOTE These following variables will be ignored if BARITO_ELASTICSEARCH_INDEX_METHOD is set to SingleInsert

• BARITO_ELASTICSEARCH_BULK_SIZE
• BARITO_ELASTICSEARCH_FLUSH_INTERVAL_MS

Changelog

See CHANGELOG.md

Contributing

See CONTRIBUTING.md

License

MIT License, See LICENSE.