From dcbaa8b336ad96c6fb7861299c81c0a9f0ee6a05 Mon Sep 17 00:00:00 2001 From: Vincenzo Date: Sun, 18 Aug 2024 20:24:39 +0200 Subject: [PATCH] feat: add cassandra producer --- go.mod | 3 + go.sum | 12 +++ pkg/cmd/producerList.go | 1 + pkg/cmd/templateRun.go | 7 +- pkg/configuration/configuration.go | 1 + pkg/emitter/emitter.go | 13 +++ pkg/producers/cassandra/config.go | 30 ++++++ pkg/producers/cassandra/config.json.example | 10 ++ pkg/producers/cassandra/producer.go | 113 ++++++++++++++++++++ 9 files changed, 188 insertions(+), 2 deletions(-) create mode 100644 pkg/producers/cassandra/config.go create mode 100644 pkg/producers/cassandra/config.json.example create mode 100644 pkg/producers/cassandra/producer.go diff --git a/go.mod b/go.mod index 776ef0a0..a69800d0 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,7 @@ require ( github.com/elastic/go-elasticsearch/v8 v8.14.0 github.com/go-chi/chi/v5 v5.1.0 github.com/go-resty/resty/v2 v2.14.0 + github.com/gocql/gocql v1.6.0 github.com/google/go-cmp v0.6.0 github.com/google/uuid v1.6.0 github.com/gorilla/sessions v1.3.0 @@ -66,6 +67,7 @@ require ( github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect github.com/googleapis/gax-go/v2 v2.12.5 // indirect github.com/gorilla/securecookie v1.1.2 // indirect + github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed // indirect github.com/hamba/avro/v2 v2.20.1 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect github.com/hashicorp/go-cleanhttp v0.5.2 // indirect @@ -119,6 +121,7 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20240624140628-dc46fd24d27d // indirect google.golang.org/grpc v1.64.0 // indirect google.golang.org/protobuf v1.34.2 // indirect + gopkg.in/inf.v0 v0.9.1 // indirect ) require ( diff --git a/go.sum b/go.sum index 2ae99145..56af2af1 100644 --- a/go.sum +++ b/go.sum @@ -84,6 +84,10 @@ github.com/bahlo/generic-list-go v0.2.0/go.mod h1:2KvAjgMlE5NNynlg/5iLrrCCZ2+5xW github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs= +github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932 h1:mXoPYz/Ul5HYEDvkta6I8/rnYM5gSdSV2tJ6XbZuEtY= +github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932/go.mod h1:NOuUCSz6Q9T7+igc/hlvDOUdtWKryOrtFyIVABv/p7k= +github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY= +github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= @@ -199,6 +203,8 @@ github.com/go-test/deep v1.0.2 h1:onZX1rnHT3Wv6cqNgYyFOOlgVKJrksuCMCRvJStbMYw= github.com/go-test/deep v1.0.2/go.mod h1:wGDj63lr65AM2AQyKZd/NYHGb0R+1RLqB8NKt3aSFNA= github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU= github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I= +github.com/gocql/gocql v1.6.0 h1:IdFdOTbnpbd0pDhl4REKQDM+Q0SzKXQ1Yh+YZZ8T/qU= +github.com/gocql/gocql v1.6.0/go.mod h1:3gM2c4D3AnkISwBxGnMMsS8Oy4y2lhbPRsH4xnJrHG8= github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= github.com/gofrs/flock v0.8.1 h1:+gYjHKf32LDeiEEFhQaotPbLuUXjY5ZqxKgXy7n59aw= github.com/gofrs/flock v0.8.1/go.mod h1:F1TvTiK9OcQqauNUHlbJvyl9Qa1QvF/gOUDKA14jxHU= @@ -224,6 +230,7 @@ github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QD github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM= github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/cel-go v0.20.1 h1:nDx9r8S3L4pE61eDdt8igGj8rf5kjYR3ILxWIpWNi84= @@ -267,6 +274,8 @@ github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 h1:+9834+KizmvFV7pXQGSXQTsaW github.com/grpc-ecosystem/go-grpc-middleware v1.3.0/go.mod h1:z0ButlSOZa5vEBq9m2m2hlwIgKw+rp3sdCBRoJY+30Y= github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 h1:YBftPWNWd4WwGqtY2yeZL2ef8rHAxPBD8KFhJpmcqms= github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0/go.mod h1:YN5jB8ie0yfIUg6VvR9Kz84aCaG7AsGZnLjhHbUqwPg= +github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed h1:5upAirOpQc1Q53c0bnx2ufif5kANL7bfZWcc6VJWJd8= +github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed/go.mod h1:tMWxXQ9wFIaZeTI9F+hmhFiGpFmhOHzyShyFUhRm0H4= github.com/hamba/avro/v2 v2.20.1 h1:3WByQiVn7wT7d27WQq6pvBRC00FVOrniP6u67FLA/2E= github.com/hamba/avro/v2 v2.20.1/go.mod h1:xHiKXbISpb3Ovc809XdzWow+XGTn+Oyf/F9aZbTLAig= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= @@ -321,8 +330,11 @@ github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 h1:Z9n2FFNU github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8= github.com/klauspost/compress v1.17.7 h1:ehO88t2UGzQK66LMdE8tibEd1ErmzZjNEqWkjLAKQQg= github.com/klauspost/compress v1.17.7/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= diff --git a/pkg/cmd/producerList.go b/pkg/cmd/producerList.go index ce4e5704..fdbc3bc0 100644 --- a/pkg/cmd/producerList.go +++ b/pkg/cmd/producerList.go @@ -54,6 +54,7 @@ var producerListCmd = &cobra.Command{ fmt.Printf("%sElastic%s (--output = elastic)\n", Green, Reset) fmt.Printf("%sS3%s (--output = s3)\n", Green, Reset) fmt.Printf("%sGCS%s (--output = gcs)\n", Green, Reset) + fmt.Printf("%sCassandr%s (--output = cassandra)\n", Green, Reset) fmt.Println() }, diff --git a/pkg/cmd/templateRun.go b/pkg/cmd/templateRun.go index 4560d48c..fa51a609 100644 --- a/pkg/cmd/templateRun.go +++ b/pkg/cmd/templateRun.go @@ -35,8 +35,8 @@ import ( var templateRunCmd = &cobra.Command{ Use: "run [template]", Short: "Execute a template", - Long: `Execute a template. - Without any other flag, [template] is just the name of a template in the templates directory, which is '$JR_SYSTEM_DIR/templates'. Example: + Long: `Execute a template. + Without any other flag, [template] is just the name of a template in the templates directory, which is '$JR_SYSTEM_DIR/templates'. Example: jr template run net_device With the --embedded flag, [template] is a string containing a full template. Example: jr template run --template "{{name}}" @@ -113,6 +113,8 @@ jr template run --template "{{name}}" configuration.GlobalCfg.GCSConfig, _ = cmd.Flags().GetString(f.Name) case "httpConfig": configuration.GlobalCfg.HTTPConfig, _ = cmd.Flags().GetString(f.Name) + case "cassandraConfig": + configuration.GlobalCfg.CassandraConfig, _ = cmd.Flags().GetString(f.Name) } } }) @@ -176,5 +178,6 @@ func init() { templateRunCmd.Flags().String("elasticConfig", "", "Elastic Search configuration") templateRunCmd.Flags().String("s3Config", "", "Amazon S3 configuration") templateRunCmd.Flags().String("gcsConfig", "", "Google GCS configuration") + templateRunCmd.Flags().String("cassandraConfig", "", "Cassandra configuration") } diff --git a/pkg/configuration/configuration.go b/pkg/configuration/configuration.go index b256c622..34c32ade 100644 --- a/pkg/configuration/configuration.go +++ b/pkg/configuration/configuration.go @@ -38,6 +38,7 @@ type GlobalConfiguration struct { S3Config string GCSConfig string HTTPConfig string + CassandraConfig string Url string EmbeddedTemplate bool FileNameTemplate bool diff --git a/pkg/emitter/emitter.go b/pkg/emitter/emitter.go index 2850862f..34a48845 100644 --- a/pkg/emitter/emitter.go +++ b/pkg/emitter/emitter.go @@ -30,6 +30,7 @@ import ( "github.com/ugol/jr/pkg/constants" "github.com/ugol/jr/pkg/ctx" "github.com/ugol/jr/pkg/functions" + "github.com/ugol/jr/pkg/producers/cassandra" "github.com/ugol/jr/pkg/producers/console" "github.com/ugol/jr/pkg/producers/elastic" "github.com/ugol/jr/pkg/producers/gcs" @@ -140,6 +141,11 @@ func (e *Emitter) Initialize(conf configuration.GlobalConfiguration) { return } + if e.Output == "cassandra" { + e.Producer = createCassandraProducer(conf.CassandraConfig) + return + } + } func (e *Emitter) Run(num int, o any) { @@ -205,6 +211,13 @@ func createHTTPProducer(httpConfig string) Producer { return httpProducer } +func createCassandraProducer(config string) Producer { + producer := &cassandra.Producer{} + producer.Initialize(config) + + return producer +} + func createKafkaProducer(conf configuration.GlobalConfiguration, topic string, templateType string) *kafka.KafkaManager { kManager := &kafka.KafkaManager{ diff --git a/pkg/producers/cassandra/config.go b/pkg/producers/cassandra/config.go new file mode 100644 index 00000000..136e5556 --- /dev/null +++ b/pkg/producers/cassandra/config.go @@ -0,0 +1,30 @@ +// Copyright © 2022 Vincenzo Marchese +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. +package cassandra + +type Config struct { + Hosts []string `json:"hosts"` + Timeout string `json:"timeout"` + Keyspace string `json:"keyspace"` + Table string `json:"table"` + ConsistencyLevel string `json:"consistencyLevel"` + Username string `json:"username"` + Password string `json:"password"` +} diff --git a/pkg/producers/cassandra/config.json.example b/pkg/producers/cassandra/config.json.example new file mode 100644 index 00000000..cbb3303f --- /dev/null +++ b/pkg/producers/cassandra/config.json.example @@ -0,0 +1,10 @@ +{ + "hosts":["host1:port1","host2:port2",...], + "keyspace": "", + "table":"", + "username": "", + "password": "", + "timeout": "", + "consistencyLevel": "" + +} diff --git a/pkg/producers/cassandra/producer.go b/pkg/producers/cassandra/producer.go new file mode 100644 index 00000000..10c2f9f6 --- /dev/null +++ b/pkg/producers/cassandra/producer.go @@ -0,0 +1,113 @@ +// Copyright © 2022 Vincenzo Marchese +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. +package cassandra + +import ( + "encoding/json" + "fmt" + "os" + "time" + + "github.com/gocql/gocql" + "github.com/rs/zerolog/log" +) + +type Producer struct { + configuration Config + + session *gocql.Session + consistencyLevel gocql.Consistency + timeout time.Duration +} + +func (p *Producer) Initialize(configFile string) { + cfgBytes, err := os.ReadFile(configFile) + if err != nil { + log.Fatal().Err(err).Msg("Failed to read config file") + } + + config := Config{} + if err := json.Unmarshal(cfgBytes, &config); err != nil { + log.Fatal().Err(err).Msg("Failed to unmarshal config") + } + + if config.Keyspace == "" { + log.Fatal().Msg("Keyspace is required") + } + + if len(config.Hosts) == 0 { + log.Fatal().Msg("Hosts are required") + } + + if config.Username == "" || config.Password == "" { + log.Fatal().Msg("Username and password are both required") + } + + if config.ConsistencyLevel == "" { + config.ConsistencyLevel = "QUORUM" + } + + consistencyLevel, err := gocql.MustParseConsistency(config.ConsistencyLevel) + if err != nil { + log.Fatal().Err(err).Msg("Failed to parse consistency level") + } + + if config.Timeout == "" { + config.Timeout = "10s" + } + + timeout, err := time.ParseDuration(config.Timeout) + if err != nil { + log.Warn().Err(err).Msg("Failed to parse timeout, setting default to 10s") + timeout = time.Second * 10 + } + + cluster := gocql.NewCluster(config.Hosts...) + cluster.Authenticator = gocql.PasswordAuthenticator{ + Username: config.Username, + Password: config.Password, + } + session, err := cluster.CreateSession() + if err != nil { + log.Fatal().Err(err).Msg("Failed to create session") + } + + p.configuration = config + p.timeout = timeout + p.session = session + p.consistencyLevel = consistencyLevel + +} + +func (p *Producer) Produce(_ []byte, v []byte, _ any) { + + stmt := fmt.Sprintf("INSERT INTO %s.%s JSON ?", + p.configuration.Keyspace, + p.configuration.Table) + if err := p.session.Query(stmt, string(v)). + Consistency(p.consistencyLevel).Exec(); err != nil { + log.Fatal().Err(err).Msg("Failed to execute query") + } +} + +func (p *Producer) Close() error { + p.session.Close() + return nil +}