feat(producer): add ability to produce to kafka (via sarama)

.gitignore

@@ -1,3 +1,6 @@
+# If you prefer the allow list template instead of the deny list, see community template:
+# https://github.com/github/gitignore/blob/main/community/Golang/Go.AllowList.gitignore
+#
 # Binaries for programs and plugins
 *.exe
 *.exe~
@@ -14,4 +17,96 @@
 # Dependency directories (remove the comment below to include it)
 # vendor/
 
-.idea
+# Go workspace file
+go.work
+
+.vscode/*
+!.vscode/settings.json
+!.vscode/tasks.json
+!.vscode/launch.json
+!.vscode/extensions.json
+!.vscode/*.code-snippets
+
+# Local History for Visual Studio Code
+.history/
+
+# Built Visual Studio Code Extensions
+*.vsix
+
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio, WebStorm and Rider
+# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
+
+# User-specific stuff
+.idea/**/workspace.xml
+.idea/**/tasks.xml
+.idea/**/usage.statistics.xml
+.idea/**/dictionaries
+.idea/**/shelf
+
+# AWS User-specific
+.idea/**/aws.xml
+
+# Generated files
+.idea/**/contentModel.xml
+
+# Sensitive or high-churn files
+.idea/**/dataSources/
+.idea/**/dataSources.ids
+.idea/**/dataSources.local.xml
+.idea/**/sqlDataSources.xml
+.idea/**/dynamic.xml
+.idea/**/uiDesigner.xml
+.idea/**/dbnavigator.xml
+
+# Gradle
+.idea/**/gradle.xml
+.idea/**/libraries
+
+# Gradle and Maven with auto-import
+# When using Gradle or Maven with auto-import, you should exclude module files,
+# since they will be recreated, and may cause churn.  Uncomment if using
+# auto-import.
+# .idea/artifacts
+# .idea/compiler.xml
+# .idea/jarRepositories.xml
+# .idea/modules.xml
+# .idea/*.iml
+# .idea/modules
+# *.iml
+# *.ipr
+
+# CMake
+cmake-build-*/
+
+# Mongo Explorer plugin
+.idea/**/mongoSettings.xml
+
+# File-based project format
+*.iws
+
+# IntelliJ
+out/
+
+# mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# JIRA plugin
+atlassian-ide-plugin.xml
+
+# Cursive Clojure plugin
+.idea/replstate.xml
+
+# SonarLint plugin
+.idea/sonarlint/
+
+# Crashlytics plugin (for Android Studio and IntelliJ)
+com_crashlytics_export_strings.xml
+crashlytics.properties
+crashlytics-build.properties
+fabric.properties
+
+# Editor-based Rest Client
+.idea/httpRequests
+
+# Android studio 3.1+ serialized cache file
+.idea/caches/build_file_checksums.ser

aconfig/config.go

@@ -2,7 +2,7 @@ package aconfig
 
 import (
 	"encoding/json"
-	"io/ioutil"
+	"os"
 	"time"
 
 	"github.com/ghodss/yaml"
@@ -62,7 +62,7 @@ type PreProcessFunc func([]byte) ([]byte, error)
 
 func New(configFile string, funcs ...PreProcessFunc) (*Application, error) {
 	var application = &Application{}
-	data, err := ioutil.ReadFile(configFile)
+	data, err := os.ReadFile(configFile)
 	if err != nil {
 		return nil, err
 	}

aconfig/custom_config.go

@@ -1,6 +1,8 @@
 package aconfig
 
-import "strings"
+import (
+	"strings"
+)
 
 const (
 	defaultLogLevel     = "info"
@@ -17,16 +19,18 @@ const (
 )
 
 type Common struct {
-	Log       Log       `json:"log,omitempty"`
-	Database  Database  `json:"database,omitempty"`
-	Encryptor Encryptor `json:"encryptor,omitempty"`
-	Var       Var       `json:"var,omitempty"`
+	Log          Log          `json:"log,omitempty"`
+	Database     Database     `json:"database,omitempty"`
+	Encryptor    Encryptor    `json:"encryptor,omitempty"`
+	Var          Var          `json:"var,omitempty"`
+	SaramaConfig SaramaConfig `json:"sarama_config,omitempty"`
 }
 
 func (c *Common) Complete(applicationName string) {
 	c.Log.complete(applicationName)
 	c.Database.complete()
 	c.Var.complete(applicationName)
+	c.SaramaConfig.complete()
 }
 
 type Var struct {

aconfig/sarama_config.go

@@ -0,0 +1,145 @@
+package aconfig
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"os"
+
+	"github.com/Shopify/sarama"
+	"github.com/dartagnanli/alpha/alog"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+// type RequiredAcks int16
+
+const (
+	// config for sarama
+	defaultKafkaClientID          = "sarama"
+	defaultKafkaChannelBufferSize = 256
+
+	// NoResponse doesn't send any response, the TCP ACK is all you get.
+	NoResponse sarama.RequiredAcks = 0
+	// WaitForLocal waits for only the local commit to succeed before responding.
+	WaitForLocal sarama.RequiredAcks = 1
+	// WaitForAll waits for all in-sync replicas to commit before responding.
+	// The minimum number of in-sync replicas is configured on the broker via
+	// the `min.insync.replicas` configuration key.
+	WaitForAll sarama.RequiredAcks = -1
+)
+
+type SaramaConfig struct {
+	ClientID                string              `json:"client_id,omitempty"`
+	ChannelBufferSize       int                 `json:"channel_buffer_size,omitempty"`
+	ProducerReturnSuccesses bool                `json:"producer_return_successes,omitempty"`
+	ProducerRetryMax        int                 `json:"producer_retry_max,omitempty"`
+	ProducerRequiredAcks    sarama.RequiredAcks `json:"producer_required_acks,omitempty"`
+	ProducerTimeout         int                 `json:"producer_timeout,omitempty"`
+	ProducerFlushFrequency  int                 `json:"producer_flush_frequency,omitempty"`
+
+	NetTLSEnable       bool        `json:"net_tls_enable,omitempty"`
+	NetTLSConfig       *tls.Config `json:"net_tls_config,omitempty"`
+	InsecureSkipVerify bool        `json:"insecure_skip_verify,omitempty"`
+	CertFile           string      `json:"cert_file,omitempty"`
+	KeyFile            string      `json:"key_file,omitempty"`
+	CaFile             string      `json:"ca_file,omitempty"`
+	K8sNamespace       string      `json:"k8s_namespace,omitempty"`
+	K8sSecret          string      `json:"k8s_secret,omitempty"`
+}
+
+func (sc *SaramaConfig) complete() {
+
+	if sc.ClientID == "" {
+		sc.ClientID = defaultKafkaClientID
+	}
+	if sc.ChannelBufferSize == 0 {
+		sc.ChannelBufferSize = defaultKafkaChannelBufferSize
+	}
+	if !sc.ProducerReturnSuccesses {
+		sc.ProducerReturnSuccesses = false
+	}
+	if sc.ProducerRetryMax == 0 {
+		sc.ProducerRetryMax = 3
+	}
+	if sc.ProducerRequiredAcks == 0 {
+		sc.ProducerRequiredAcks = WaitForLocal
+	}
+	if sc.InsecureSkipVerify {
+		sc.InsecureSkipVerify = true
+	}
+	if sc.NetTLSEnable {
+		tlsConfig := sc.createTLSConfigurationWithK8s()
+		if tlsConfig != nil {
+			sc.NetTLSConfig = tlsConfig
+		}
+	}
+	if sc.ProducerTimeout == 0 {
+		sc.ProducerTimeout = 10
+	}
+	if sc.ProducerFlushFrequency == 0 {
+		sc.ProducerFlushFrequency = 500
+	}
+}
+
+func (sc *SaramaConfig) createTLSConfiguration() (t *tls.Config) {
+	if sc.CertFile != "" && sc.KeyFile != "" && sc.CaFile != "" {
+		cert, err := tls.LoadX509KeyPair(sc.CertFile, sc.KeyFile)
+		if err != nil {
+			panic(err)
+		}
+
+		caCert, err := os.ReadFile(sc.CaFile)
+		if err != nil {
+			panic(err)
+		}
+
+		caCertPool := x509.NewCertPool()
+		caCertPool.AppendCertsFromPEM(caCert)
+
+		t = &tls.Config{
+			Certificates:       []tls.Certificate{cert},
+			RootCAs:            caCertPool,
+			InsecureSkipVerify: sc.InsecureSkipVerify,
+		}
+	}
+	// will be nil by default if nothing is provided
+	return t
+}
+
+func (sc *SaramaConfig) createTLSConfigurationWithK8s() (t *tls.Config) {
+
+	cfg, err := rest.InClusterConfig()
+	if err != nil {
+		panic(err.Error())
+	}
+	clientset, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		panic(err.Error())
+	}
+
+	secret := sc.K8sSecret
+	namespace := sc.K8sNamespace
+	mySecret, err := clientset.CoreV1().Secrets(namespace).Get(context.TODO(), secret, metav1.GetOptions{})
+	if err != nil {
+		if errors.IsNotFound(err) {
+			alog.Sugar.Errorf("Secret %s in namespace %s not found\n", secret, namespace)
+		}
+	}
+	cert, err := tls.X509KeyPair(mySecret.Data["cert.pem"], mySecret.Data["key.pem"])
+	if err != nil {
+		panic(err.Error())
+	}
+	caCertPool := x509.NewCertPool()
+	caCertPool.AppendCertsFromPEM(mySecret.Data["ca_cert.pem"])
+
+	t = &tls.Config{
+		Certificates:       []tls.Certificate{cert},
+		RootCAs:            caCertPool,
+		InsecureSkipVerify: sc.InsecureSkipVerify,
+	}
+
+	return t
+}

alog/alog.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/alphaframework/alpha/autil"
-	"github.com/alphaframework/alpha/autil/ahttp/request"
+	"github.com/dartagnanli/alpha/autil"
+	"github.com/dartagnanli/alpha/autil/ahttp/request"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"gopkg.in/natefinch/lumberjack.v2"

alog/gormwrapper/logger.go

@@ -5,11 +5,11 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/dartagnanli/alpha/alog"
+	"github.com/dartagnanli/alpha/autil/ahttp/request"
 	"go.uber.org/zap"
 	gormlogger "gorm.io/gorm/logger"
 	"gorm.io/gorm/utils"
-	"github.com/alphaframework/alpha/alog"
-	"github.com/alphaframework/alpha/autil/ahttp/request"
 )
 
 type Config struct {

autil/acrypto/pbe/pbe.go

@@ -5,7 +5,8 @@ import (
 	"crypto/md5"
 	"crypto/rand"
 	"encoding/base64"
-	"github.com/alphaframework/alpha/autil/acrypto"
+
+	"github.com/dartagnanli/alpha/autil/acrypto"
 )
 
 func getDerivedKey(password string, salt []byte, count, l int) ([]byte, error) {

autil/ahttp/request/context.go

@@ -2,6 +2,7 @@ package request
 
 import (
 	"context"
+
 	"github.com/gin-gonic/gin"
 )
 

database/database.go

@@ -5,12 +5,12 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/dartagnanli/alpha/aconfig"
+	"github.com/dartagnanli/alpha/alog"
+	"github.com/dartagnanli/alpha/alog/gormwrapper"
 	"gorm.io/driver/mysql"
 	"gorm.io/gorm"
 	gormlogger "gorm.io/gorm/logger"
-	"github.com/alphaframework/alpha/aconfig"
-	"github.com/alphaframework/alpha/alog"
-	"github.com/alphaframework/alpha/alog/gormwrapper"
 )
 
 func MustNewDBWith(portName aconfig.PortName, appConfig *aconfig.Application, driver string, commonConfig *aconfig.Database) *gorm.DB {

examples/main.go

@@ -3,17 +3,17 @@ package main
 import (
 	"fmt"
 
-	_ "github.com/alphaframework/alpha/aconfig"
-	_ "github.com/alphaframework/alpha/aerror"
-	_ "github.com/alphaframework/alpha/alog"
-	_ "github.com/alphaframework/alpha/alog/gormwrapper"
-	_ "github.com/alphaframework/alpha/autil"
-	_ "github.com/alphaframework/alpha/autil/ahttp"
-	_ "github.com/alphaframework/alpha/autil/ahttp/request"
-	_ "github.com/alphaframework/alpha/database"
-	_ "github.com/alphaframework/alpha/ginwrapper"
-	_ "github.com/alphaframework/alpha/httpclient"
-	_ "github.com/alphaframework/alpha/httpserver/rsp"
+	_ "github.com/dartagnanli/alpha/aconfig"
+	_ "github.com/dartagnanli/alpha/aerror"
+	_ "github.com/dartagnanli/alpha/alog"
+	_ "github.com/dartagnanli/alpha/alog/gormwrapper"
+	_ "github.com/dartagnanli/alpha/autil"
+	_ "github.com/dartagnanli/alpha/autil/ahttp"
+	_ "github.com/dartagnanli/alpha/autil/ahttp/request"
+	_ "github.com/dartagnanli/alpha/database"
+	_ "github.com/dartagnanli/alpha/ginwrapper"
+	_ "github.com/dartagnanli/alpha/httpclient"
+	_ "github.com/dartagnanli/alpha/httpserver/rsp"
 )
 
 func main() {

ginwrapper/ginwrapper.go

@@ -3,7 +3,7 @@ package ginwrapper
 import (
 	"time"
 
-	"github.com/alphaframework/alpha/alog"
+	"github.com/dartagnanli/alpha/alog"
 	"github.com/gin-gonic/gin"
 )