feat: add additional support for running locally

README.md

@@ -163,6 +163,31 @@ Batch log entries endpoint:
 }
 ```
 
+### gRPC API
+Use [barito-proto](https://github.com/bentol/barito-proto/blob/master/producer/producer.proto) protocol or enable reflection in configuration.
+
+Produce single log using [gRPCurl](https://github.com/fullstorydev/grpcurl):
+``` 
+grpcurl -plaintext -d @ localhost:8082 producer.Producer/Produce << EOF
+{ 
+  "context": { 
+    "kafka_topic": "kafka_topic", 
+    "kafka_partition": 1, 
+    "kafka_replication_factor": 1, 
+    "es_index_prefix": "test", 
+    "es_document_type":"test", 
+    "app_max_tps":100, 
+    "app_secret":"app_secret" 
+  },
+  "content": { 
+    "hello": "world", 
+    "key": "value", 
+    "num": 100 
+  } 
+}
+EOF
+```
+
 ### Producer Configuration
 
 These environment variables can be modified to customize producer behavior:
@@ -180,6 +205,7 @@ These environment variables can be modified to customize producer behavior:
 | ProducerMaxRetry | Set kafka setting max retry | BARITO_PRODUCER_MAX_RETRY | 10 |
 | ProducerMaxTps | Producer rate limit trx per second | BARITO_PRODUCER_MAX_TPS | 100 |
 | ProducerRateLimitResetInterval | Producer rate limit reset interval (in seconds) | BARITO_PRODUCER_RATE_LIMIT_RESET_INTERVAL | 10 |
+| GrpcWithReflection | Toggle for using reflection in gRPC | BARITO_GRPC_WITH_REFLECTION | false |
 
 ## Consumer Mode
 

cmds/action.go

@@ -129,6 +129,7 @@ func ActionBaritoProducerService(c *cli.Context) (err error) {
 	kafkaRetryInterval := configKafkaRetryInterval()
 	newTopicEventName := configNewTopicEvent()
 	grpcMaxRecvMsgSize := configGrpcMaxRecvMsgSize()
+	grpcWithReflection := configGrpcWithReflection()
 	rateLimiterOpt := configRateLimiterOpt()
 	maxMessageBytes := configProducerMaxMessageBytes()
 	kafkaMessageFormat := configKafkaMessageFormat()
@@ -185,6 +186,7 @@ func ActionBaritoProducerService(c *cli.Context) (err error) {
 		"kafkaRetryInterval": kafkaRetryInterval,
 		"newEventTopic":      newTopicEventName,
 		"grpcMaxRecvMsgSize": grpcMaxRecvMsgSize,
+		"grpcWithReflection": grpcWithReflection,
 		"ignoreKafkaOptions": ignoreKafkaOptions,
 		"limiter":            rateLimiter,
 		"kafkaMessageFormat": kafkaMessageFormat,
@@ -285,14 +287,21 @@ func setupRedisRateLimiter(_ context.Context,
 func setupRedactor() *redact.Redactor {
 	var redactor *redact.Redactor
 	var err error
-	if redactorRulesMap := configRedactorRulesMap(); redactorRulesMap == "" {
-		marketEndpoint := configMarketRedactUrl()
+	if marketEndpoint := configMarketRedactUrl(); marketEndpoint == "" {
+		redactorRulesMap := configRedactorRulesMap()
+		redactor, err = redact.NewRedactorFromJSON(redactorRulesMap)
+		if err != nil {
+			log.Errorf("Error creating redactor from JSON rules map: %s", err)
+			return nil
+		}
+	} else {
 		clusterName := configClusterName()
 		marketClientKey := configMarketClientKey()
 		redactor, err = redact.NewRedactorFromMarket(marketEndpoint, clusterName, marketClientKey)
 		if err != nil {
 			return nil
 		}
 	}
+
 	return redactor
 }

cmds/config.go

@@ -25,6 +25,7 @@ const (
 	EnvEsDatastreamDefaultComponentTemplateName = "BARITO_ELASTICSEARCH_DATASTREAM_DEFAULT_COMPONENT_TEMPLATE_NAME"
 
 	EnvGrpcMaxRecvMsgSize = "BARITO_GRPC_MAX_RECV_MSG_SIZE"
+	EnvGrpcWithReflection = "BARITO_GRPC_WITH_REFLECTION"
 
 	EnvPushMetricUrl   = "BARITO_PUSH_METRIC_URL"
 	EnvMarketRedactUrl = "BARITO_MARKET_REDACT_ENDPOINT_URL"
@@ -89,6 +90,7 @@ var (
 	DefaultElasticsearchUrls = []string{"http://localhost:9200"}
 
 	DefaultGrpcMaxRecvMsgSize = 20 * 1000 * 1000
+	DefaultGrpcWithReflection = "false"
 
 	DefaultPushMetricUrl   = ""
 	DefaultMarketRedactUrl = ""
@@ -128,6 +130,8 @@ var (
 	DefaultRedisUrl       = "http://localhost:6379"
 	DefaultRedisPassword  = ""
 	DefaultRedisKeyPrefix = "barito:producer:ratelimit:"
+
+	DefaultRedactorRulesMap = "{}"
 )
 
 func configKafkaBrokers() (brokers []string) {
@@ -189,6 +193,10 @@ func configGrpcMaxRecvMsgSize() (i int) {
 	return intEnvOrDefault(EnvGrpcMaxRecvMsgSize, DefaultGrpcMaxRecvMsgSize)
 }
 
+func configGrpcWithReflection() bool {
+	return (stringEnvOrDefault(EnvGrpcWithReflection, DefaultGrpcWithReflection) == "true")
+}
+
 func configConsulElasticsearchName() (s string) {
 	return stringEnvOrDefault(EnvConsulElasticsearchName, DefaultConsulElasticsearchName)
 }
@@ -343,7 +351,7 @@ func configRateLimiterOpt() RateLimiterOpt {
 }
 
 func configRedactorRulesMap() (s string) {
-	return stringEnvOrDefault(EnvRedactorRulesMap, "")
+	return stringEnvOrDefault(EnvRedactorRulesMap, DefaultRedactorRulesMap)
 }
 
 func configRedisUrl() (url string) {

docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM --platform=arm64 golang:1.22.3-alpine as builder
+FROM --platform=arm64 golang:1.24.10-alpine3.22 as builder
 
 WORKDIR /go/src/github.com/BaritoLog/barito-flow
 COPY . .

docker/docker-compose.yml

@@ -2,16 +2,40 @@ version: "3.7"
 
 services:
   es:
-    image: docker.elastic.co/elasticsearch/elasticsearch:6.3.2
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.17.28
+    environment:
+      - discovery.type=single-node
     ports:
       - 9200:9200
       - 9300:9300
-
-  kafka:
-    image: spotify/kafka
+  kibana:
+    image: docker.elastic.co/kibana/kibana:7.17.28
+    environment:
+      - ELASTICSEARCH_HOSTS=http://es:9200
     ports:
-      - 9092:9092
+      - "5601:5601"
+    depends_on:
+      - es
+  zookeeper:
+    image: confluentinc/cp-zookeeper:3.2.0
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+    ports: 
       - 2181:2181
+  kafka:
+    depends_on:
+      - zookeeper
+    image: quay.io/strimzi/kafka:0.31.1-kafka-3.2.0
+    command: [
+      "sh", "-c",
+      "./bin/kafka-server-start.sh /tmp/server.properties"
+    ]
+    ports:
+      - "9092:9092"
+    environment:
+      LOG_DIR: "/tmp/logs"
+    volumes:
+      - ./server.properties:/tmp/server.properties:ro
   redis:
     image: redis:alpine3.16
     ports:
@@ -27,7 +51,7 @@ services:
     environment:
       - BARITO_KAFKA_BROKERS=kafka:9092
     ports:
-      - 8080:8080
+      - 8082:8082
   consumer:
     image: baritolog/barito-flow
     build:

docker/server.properties

@@ -0,0 +1,138 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#
+# This configuration file is intended for use in ZK-based mode, where Apache ZooKeeper is required.
+# See kafka.server.KafkaConfig for additional details and defaults
+#
+
+############################# Server Basics #############################
+
+# The id of the broker. This must be set to a unique integer for each broker.
+broker.id=0
+
+############################# Socket Server Settings #############################
+
+# The address the socket server listens on. If not configured, the host name will be equal to the value of
+# java.net.InetAddress.getCanonicalHostName(), with PLAINTEXT listener name, and port 9092.
+#   FORMAT:
+#     listeners = listener_name://host_name:port
+#   EXAMPLE:
+#     listeners = PLAINTEXT://your.host.name:9092
+listeners=PLAINTEXT://:9092
+
+# Listener name, hostname and port the broker will advertise to clients.
+# If not set, it uses the value for "listeners".
+advertised.listeners=PLAINTEXT://kafka:9092
+
+# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
+#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
+
+# The number of threads that the server uses for receiving requests from the network and sending responses to the network
+num.network.threads=3
+
+# The number of threads that the server uses for processing requests, which may include disk I/O
+num.io.threads=8
+
+# The send buffer (SO_SNDBUF) used by the socket server
+socket.send.buffer.bytes=102400
+
+# The receive buffer (SO_RCVBUF) used by the socket server
+socket.receive.buffer.bytes=102400
+
+# The maximum size of a request that the socket server will accept (protection against OOM)
+socket.request.max.bytes=104857600
+
+
+############################# Log Basics #############################
+
+# A comma separated list of directories under which to store log files
+log.dirs=/tmp/kafka-logs
+
+# The default number of log partitions per topic. More partitions allow greater
+# parallelism for consumption, but this will also result in more files across
+# the brokers.
+num.partitions=1
+
+# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
+# This value is recommended to be increased for installations with data dirs located in RAID array.
+num.recovery.threads.per.data.dir=1
+
+############################# Internal Topic Settings  #############################
+# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
+# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
+offsets.topic.replication.factor=1
+transaction.state.log.replication.factor=1
+transaction.state.log.min.isr=1
+
+############################# Log Flush Policy #############################
+
+# Messages are immediately written to the filesystem but by default we only fsync() to sync
+# the OS cache lazily. The following configurations control the flush of data to disk.
+# There are a few important trade-offs here:
+#    1. Durability: Unflushed data may be lost if you are not using replication.
+#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
+#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
+# The settings below allow one to configure the flush policy to flush data after a period of time or
+# every N messages (or both). This can be done globally and overridden on a per-topic basis.
+
+# The number of messages to accept before forcing a flush of data to disk
+#log.flush.interval.messages=10000
+
+# The maximum amount of time a message can sit in a log before we force a flush
+#log.flush.interval.ms=1000
+
+############################# Log Retention Policy #############################
+
+# The following configurations control the disposal of log segments. The policy can
+# be set to delete segments after a period of time, or after a given size has accumulated.
+# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
+# from the end of the log.
+
+# The minimum age of a log file to be eligible for deletion due to age
+log.retention.hours=168
+
+# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
+# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
+#log.retention.bytes=1073741824
+
+# The maximum size of a log segment file. When this size is reached a new log segment will be created.
+log.segment.bytes=1073741824
+
+# The interval at which log segments are checked to see if they can be deleted according
+# to the retention policies
+log.retention.check.interval.ms=300000
+
+############################# Zookeeper #############################
+
+# Zookeeper connection string (see zookeeper docs for details).
+# This is a comma separated host:port pairs, each corresponding to a zk
+# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
+# You can also append an optional chroot string to the urls to specify the
+# root directory for all kafka znodes.
+zookeeper.connect=zookeeper:2181
+
+# Timeout in ms for connecting to zookeeper
+zookeeper.connection.timeout.ms=18000
+
+
+############################# Group Coordinator Settings #############################
+
+# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
+# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
+# The default value for this is 3 seconds.
+# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
+# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
+group.initial.rebalance.delay.ms=0

flow/grpc_producer.go

@@ -13,6 +13,7 @@ import (
 	pb "github.com/bentol/barito-proto/producer"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/reflection"
 
 	_ "github.com/mostynb/go-grpc-compression/zstd"
 )
@@ -45,6 +46,7 @@ type producerService struct {
 	kafkaRetryInterval int
 	newEventTopic      string
 	grpcMaxRecvMsgSize int
+	grpcWithReflection bool
 	ignoreKafkaOptions bool
 	kafkaMessageFormat string
 
@@ -67,6 +69,7 @@ func NewProducerService(params map[string]interface{}) *producerService {
 		kafkaRetryInterval:          params["kafkaRetryInterval"].(int),
 		newEventTopic:               params["newEventTopic"].(string),
 		grpcMaxRecvMsgSize:          params["grpcMaxRecvMsgSize"].(int),
+		grpcWithReflection:          params["grpcWithReflection"].(bool),
 		ignoreKafkaOptions:          params["ignoreKafkaOptions"].(bool),
 		kafkaMessageFormat:          params["kafkaMessageFormat"].(string),
 		limiter:                     params["limiter"].(RateLimiter),
@@ -133,6 +136,10 @@ func (s *producerService) initGrpcServer() (lis net.Listener, srv *grpc.Server,
 	srv = grpc.NewServer(grpc.MaxRecvMsgSize(s.grpcMaxRecvMsgSize))
 	pb.RegisterProducerServer(srv, s)
 
+	if s.grpcWithReflection {
+		reflection.Register(srv)
+	}
+
 	s.grpcServer = srv
 	return
 }