Skip to content

Commit 90f6cbd

Browse files
Yashwanth-Ranjan-SingaravelRSYashwanth
Yashwanth-Ranjan-Singaravel
and
RSYashwanth
authored
Added Kafka worker-side consumer logic (#342)
* Added workerside kafka server * Changed the sarama kafka client package to kgo * Added preloading of kafka consumers * Initial code cleanup + removed preloading and coupling of lambda/ with event/ * Implemented API to register/unregister kafka consumers per worker * Code cleanup * Initial server.go changes for mux * Reverted server.go and lambdaConfig.go --------- Co-authored-by: RSYashwanth <yashwanthrs1104@gmail.com>
1 parent 0a43391 commit 90f6cbd

4 files changed

Lines changed: 395 additions & 7 deletions

File tree

‎go/boss/lambdastore/store.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -134,7 +134,7 @@ func (s *LambdaStore) DeleteLambda(w http.ResponseWriter, r *http.Request) {
134134
}
135135

136136
func (s *LambdaStore) ListLambda(w http.ResponseWriter) {
137-
funcNames := s.listEntries()
137+
funcNames := s.ListEntries()
138138

139139
if err := json.NewEncoder(w).Encode(funcNames); err != nil {
140140
http.Error(w, "failed to encode lambda list", http.StatusInternalServerError)
@@ -157,7 +157,7 @@ func (s *LambdaStore) RetrieveLambdaConfig(w http.ResponseWriter, r *http.Reques
157157
return
158158
}
159159

160-
cfg, err := s.getConfig(funcName)
160+
cfg, err := s.GetConfig(funcName)
161161
if err != nil {
162162
http.Error(w, err.Error(), http.StatusNotFound)
163163
return
@@ -310,7 +310,7 @@ func (s *LambdaStore) removeFromRegistry(funcName string) error {
310310
return nil
311311
}
312312

313-
func (s *LambdaStore) getConfig(funcName string) (*common.LambdaConfig, error) {
313+
func (s *LambdaStore) GetConfig(funcName string) (*common.LambdaConfig, error) {
314314
lambdaEntry := s.getOrCreateEntry(funcName)
315315
lambdaEntry.Lock.Lock()
316316
defer lambdaEntry.Lock.Unlock()
@@ -344,7 +344,7 @@ func (s *LambdaStore) getOrCreateEntry(funcName string) *LambdaEntry {
344344
return entry
345345
}
346346

347-
func (s *LambdaStore) listEntries() []string {
347+
func (s *LambdaStore) ListEntries() []string {
348348
s.mapLock.Lock()
349349
defer s.mapLock.Unlock()
350350

‎go/go.mod‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ require (
1010
github.com/fsouza/go-dockerclient v1.10.1
1111
github.com/golang-jwt/jwt v3.2.2+incompatible
1212
github.com/robfig/cron/v3 v3.0.1
13+
github.com/twmb/franz-go v1.19.0
1314
github.com/urfave/cli/v2 v2.25.3
1415
gocloud.dev v0.42.0
1516
gopkg.in/yaml.v3 v3.0.1
@@ -81,7 +82,7 @@ require (
8182
github.com/googleapis/enterprise-certificate-proxy v0.3.6 // indirect
8283
github.com/googleapis/gax-go/v2 v2.14.2 // indirect
8384
github.com/jmespath/go-jmespath v0.4.0 // indirect
84-
github.com/klauspost/compress v1.17.11 // indirect
85+
github.com/klauspost/compress v1.18.0 // indirect
8586
github.com/kylelemons/godebug v1.1.0 // indirect
8687
github.com/moby/patternmatcher v0.6.0 // indirect
8788
github.com/moby/sys/sequential v0.5.0 // indirect
@@ -90,12 +91,14 @@ require (
9091
github.com/morikuni/aec v1.0.0 // indirect
9192
github.com/opencontainers/go-digest v1.0.0 // indirect
9293
github.com/opencontainers/image-spec v1.1.0-rc2.0.20221005185240-3a7f492d3f1b // indirect
94+
github.com/pierrec/lz4/v4 v4.1.22 // indirect
9395
github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
9496
github.com/pkg/errors v0.9.1 // indirect
9597
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
9698
github.com/russross/blackfriday/v2 v2.1.0 // indirect
9799
github.com/sirupsen/logrus v1.9.3 // indirect
98100
github.com/spiffe/go-spiffe/v2 v2.5.0 // indirect
101+
github.com/twmb/franz-go/pkg/kmsg v1.11.2 // indirect
99102
github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
100103
github.com/zeebo/errs v1.4.0 // indirect
101104
go.opencensus.io v0.24.0 // indirect

‎go/go.sum‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -189,8 +189,8 @@ github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGw
189189
github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
190190
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
191191
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
192-
github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
193-
github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
192+
github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
193+
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
194194
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
195195
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
196196
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -211,6 +211,8 @@ github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8
211211
github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
212212
github.com/opencontainers/image-spec v1.1.0-rc2.0.20221005185240-3a7f492d3f1b h1:YWuSjZCQAPM8UUBLkYUk1e+rZcvWHJmFb6i6rM44Xs8=
213213
github.com/opencontainers/image-spec v1.1.0-rc2.0.20221005185240-3a7f492d3f1b/go.mod h1:3OVijpioIKYWTqjiG0zfF6wvoJ4fAXGbjdZuI2NgsRQ=
214+
github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=
215+
github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
214216
github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ=
215217
github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU=
216218
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -241,6 +243,10 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
241243
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
242244
github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
243245
github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
246+
github.com/twmb/franz-go v1.19.0 h1:FzBAPUeaip68X9cbLDesgQesa5zxKVaZMk+du98vj3c=
247+
github.com/twmb/franz-go v1.19.0/go.mod h1:4kFJ5tmbbl7asgwAGVuyG1ZMx0NNpYk7EqflvWfPCpM=
248+
github.com/twmb/franz-go/pkg/kmsg v1.11.2 h1:hIw75FpwcAjgeyfIGFqivAvwC5uNIOWRGvQgZhH4mhg=
249+
github.com/twmb/franz-go/pkg/kmsg v1.11.2/go.mod h1:CFfkkLysDNmukPYhGzuUcDtf46gQSqCZHMW1T4Z+wDE=
244250
github.com/urfave/cli/v2 v2.25.3 h1:VJkt6wvEBOoSjPFQvOkv6iWIrsJyCrKGtCtxXWwmGeY=
245251
github.com/urfave/cli/v2 v2.25.3/go.mod h1:GHupkWPMM0M/sj1a2b4wUrWBPzazNrIjouW6fmdJLxc=
246252
github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 h1:bAn7/zixMGCfxrRTfdpNzjtPYqr8smhKouy9mxVdGPU=

0 commit comments

Comments
 (0)