diff --git a/relay-server/Dockerfile b/relay-server/Dockerfile index 2a0a0eb..1037955 100644 --- a/relay-server/Dockerfile +++ b/relay-server/Dockerfile @@ -9,7 +9,7 @@ WORKDIR /usr/src/kubearmor-relay-server COPY . . -RUN make +RUN cd relay-server && make ### Copy executable image @@ -26,14 +26,14 @@ LABEL name="kubearmor-relay-server" \ alerts, and system logs generated by KubeArmor in each node, streamlining log integration with other systems." RUN microdnf -y update && \ - microdnf -y install --nodocs --setopt=install_weak_deps=0 --setopt=keepcache=0 shadow-utils && \ - microdnf clean all + microdnf -y install --nodocs --setopt=install_weak_deps=0 --setopt=keepcache=0 shadow-utils && \ + microdnf clean all RUN groupadd --gid 1000 default \ - && useradd --uid 1000 --gid default --shell /bin/bash --create-home default + && useradd --uid 1000 --gid default --shell /bin/bash --create-home default COPY LICENSE /licenses/license.txt -COPY --from=builder /usr/src/kubearmor-relay-server/kubearmor-relay-server /KubeArmor/kubearmor-relay-server +COPY --from=builder /usr/src/kubearmor-relay-server/relay-server/kubearmor-relay-server /KubeArmor/kubearmor-relay-server USER default diff --git a/relay-server/elasticsearch/adapter.go b/relay-server/elasticsearch/adapter.go index 2be13a8..28db4d6 100644 --- a/relay-server/elasticsearch/adapter.go +++ b/relay-server/elasticsearch/adapter.go @@ -17,6 +17,7 @@ import ( "github.com/google/uuid" kg "github.com/kubearmor/kubearmor-relay-server/relay-server/log" + pb "github.com/kubearmor/KubeArmor/protobuf" kl "github.com/kubearmor/kubearmor-relay-server/relay-server/common" // kif "github.com/kubearmor/kubearmor-relay-server/relay-server/informers" @@ -36,8 +37,8 @@ type ElasticsearchClient struct { cancel context.CancelFunc bulkIndexer esutil.BulkIndexer ctx context.Context - alertCh chan interface{} - logCh chan interface{} + alertCh chan *pb.Alert + logCh chan *pb.Log // client *kif.Client } @@ -74,8 +75,8 @@ func NewElasticsearchClient(esURL, Endpoint string) (*ElasticsearchClient, error if err != nil { log.Fatalf("Error creating the indexer: %s", err) } - alertCh := make(chan interface{}, 10000) - logCh := make(chan interface{}, 10000) + alertCh := make(chan *pb.Alert, 10000) + logCh := make(chan *pb.Log, 10000) kaClient := server.NewClient(Endpoint) // k8sClient := kif.GetK8sClient() @@ -147,7 +148,14 @@ func (ecl *ElasticsearchClient) Start() error { } kg.Printf("Checked the liveness of the gRPC server") - client.WgServer.Go(func() error { + // var wg sync.WaitGroup + + // stop := make(chan struct{}) + // errCh := make(chan error, 1) + + client.WgServer.Add(1) + go func() error { + defer client.WgServer.Done() for client.Running { res, err := client.AlertStream.Recv() if err != nil { @@ -165,8 +173,9 @@ func (ecl *ElasticsearchClient) Start() error { //not able to add it to Log buffer } } + return nil - }) + }() for i := 0; i < 5; i++ { go func() { @@ -181,8 +190,12 @@ func (ecl *ElasticsearchClient) Start() error { } }() } + client.WgServer.Add(1) + go func() error { + + defer client.WgServer.Done() + // client.WatchLogs(&wg, stop, errCh, ecl.logCh) - client.WgServer.Go(func() error { for client.Running { res, err := client.LogStream.Recv() if err != nil { @@ -223,8 +236,9 @@ func (ecl *ElasticsearchClient) Start() error { fmt.Printf("%s\n", string(tel)) ecl.logCh <- res } + return nil - }) + }() for i := 0; i < 5; i++ { go func() { diff --git a/relay-server/server/relayServer.go b/relay-server/server/relayServer.go index 65d7df6..a782a60 100644 --- a/relay-server/server/relayServer.go +++ b/relay-server/server/relayServer.go @@ -16,7 +16,6 @@ import ( "github.com/google/uuid" pb "github.com/kubearmor/KubeArmor/protobuf" - "golang.org/x/sync/errgroup" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" @@ -302,7 +301,7 @@ type LogClient struct { Ifclient *kif.Client // wait group - WgServer *errgroup.Group + WgServer sync.WaitGroup Context context.Context } @@ -384,6 +383,9 @@ func NewClient(server string) *LogClient { lc.Ifclient = Informerclient + // var g errgroup.Group + lc.WgServer = sync.WaitGroup{} + return lc } @@ -581,7 +583,6 @@ func (lc *LogClient) WatchLogs(wg *sync.WaitGroup, stop chan struct{}, errCh cha } } - return } select {