Skip to content

Commit 222c221

Browse files
matthyxmatthyx
authored
Merge pull request #145 from kubescape/disco-client
get services from API, removing sidecar requirement
2 parents 080f76e + 6e5ad9e commit 222c221

11 files changed

Lines changed: 867 additions & 570 deletions

File tree

‎.github/workflows/pr-created.yaml‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ on:
1111
env:
1212
DB_MIGRATIONS_GH_USER: ${{ secrets.DB_MIGRATIONS_GH_USER }}
1313
DB_MIGRATIONS_GH_TOKEN: ${{ secrets.DB_MIGRATIONS_GH_TOKEN }}
14+
GIT_LFS_SKIP_SMUDGE: "1"
1415

1516
concurrency:
1617
group: ${{ github.workflow }}-${{ github.ref }}
@@ -62,7 +63,12 @@ jobs:
6263
- name: Print Go version
6364
run: go version
6465

66+
- name: Checkout db-migrations
67+
run: |
68+
git clone --depth=1 https://${{ secrets.DB_MIGRATIONS_GH_USER }}:${{ secrets.DB_MIGRATIONS_GH_TOKEN }}@github.com/armosec/db-migrations.git /tmp/db-migrations
69+
echo "LOCAL_FILE_MIGRATION_PATH=/tmp/db-migrations/postgres/hot_migrations" >> $GITHUB_ENV
70+
6571
- name: Run test
6672
run: |
6773
echo "machine github.com login git password ${{ secrets.INGESTERS_READ_TOKEN }}" > ~/.netrc
68-
cd tests && go test -v ./... -run ${{ matrix.test }} --timeout=20m --tags=integration
74+
cd tests && go test -v ./... -run ${{ matrix.test }} --timeout=40m --tags=integration

‎adapters/backend/v1/pulsar.go‎

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"encoding/json"
66
"fmt"
77
"os"
8+
"strings"
89
"sync"
910
"time"
1011

@@ -19,7 +20,6 @@ import (
1920
"github.com/kubescape/synchronizer/utils"
2021
"github.com/prometheus/client_golang/prometheus"
2122
"go.uber.org/multierr"
22-
"k8s.io/utils/ptr"
2323
)
2424

2525
const (
@@ -112,6 +112,18 @@ func (c *PulsarMessageReader) readerLoop(ctx context.Context) {
112112
for {
113113
msg, err := c.reader.Next(ctx)
114114
if err != nil {
115+
// If the context was cancelled or the consumer was closed during shutdown,
116+
// return gracefully instead of calling Fatal (which would call os.Exit(1))
117+
select {
118+
case <-ctx.Done():
119+
logger.L().Ctx(ctx).Info("pulsar reader loop exiting due to context cancellation")
120+
return
121+
default:
122+
}
123+
if strings.Contains(err.Error(), "ConsumerClosed") || strings.Contains(err.Error(), "consumer closed") {
124+
logger.L().Ctx(ctx).Info("pulsar reader loop exiting due to consumer close")
125+
return
126+
}
115127
logger.L().Ctx(ctx).Fatal("failed to read message from pulsar", helpers.Error(err))
116128
}
117129

@@ -330,11 +342,10 @@ func NewPulsarMessageProducer(cfg config.Config, pulsarClient pulsarconnector.Cl
330342
fullTopic := pulsarconnector.BuildPersistentTopic(pulsarClient.GetConfig().Tenant, pulsarClient.GetConfig().Namespace, topic)
331343

332344
options := pulsar.ProducerOptions{
333-
DisableBatching: true,
334-
EnableChunking: true,
335-
CompressionType: pulsar.ZSTD,
336-
CompressionLevel: 1,
337-
MaxReconnectToBroker: ptr.To(uint(5)),
345+
DisableBatching: true,
346+
EnableChunking: true,
347+
CompressionType: pulsar.ZSTD,
348+
CompressionLevel: 1,
338349
Properties: map[string]string{
339350
"podName": os.Getenv("HOSTNAME"),
340351
},

‎adapters/httpendpoint/v1/adapter.go‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -224,9 +224,19 @@ func (a *Adapter) Start(ctx context.Context) error {
224224
}
225225
a.supportedPaths[lowerCaseStrategy][resource.Group][resource.Version][resource.Resource] = true
226226
}
227+
// Pre-bind the listener so we can return an error immediately if the port is unavailable.
228+
ln, err := net.Listen("tcp", a.httpServer.Addr)
229+
if err != nil {
230+
return fmt.Errorf("httpendpoint failed to listen on %s: %w", a.httpServer.Addr, err)
231+
}
227232
go func() {
228-
if err := a.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
229-
logger.L().Ctx(ctx).Fatal("httpendpoint server error", helpers.Error(err))
233+
if err := a.httpServer.Serve(ln); err != nil && err != http.ErrServerClosed {
234+
select {
235+
case <-ctx.Done():
236+
logger.L().Ctx(ctx).Info("httpendpoint server stopped due to context cancellation", helpers.Error(err))
237+
default:
238+
logger.L().Ctx(ctx).Error("httpendpoint server error", helpers.Error(err))
239+
}
230240
}
231241
logger.L().Ctx(ctx).Info("httpendpoint server stopped")
232242
}()

‎cmd/client/main.go‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,11 @@ func main() {
5757
}
5858

5959
// synchronizer server URL using service discovery
60-
if services, err := config.LoadServiceURLs("/etc/config/services.json"); err != nil {
60+
apiURL := os.Getenv("API_URL")
61+
if apiURL == "" {
62+
apiURL = "api.armosec.io"
63+
}
64+
if services, err := config.LoadServiceURLs(apiURL); err != nil {
6165
logger.L().Warning("failed discovering urls", helpers.Error(err))
6266
} else if serverUrl := services.GetSynchronizerUrl(); serverUrl != "" {
6367
logger.L().Debug("synchronizer server URL", helpers.String("synchronizer", serverUrl))

‎config/config.go‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import (
88
"github.com/armosec/utils-k8s-go/armometadata"
99
"github.com/kubescape/backend/pkg/servicediscovery"
1010
"github.com/kubescape/backend/pkg/servicediscovery/schema"
11-
v2 "github.com/kubescape/backend/pkg/servicediscovery/v2"
11+
v3 "github.com/kubescape/backend/pkg/servicediscovery/v3"
1212
pulsarconfig "github.com/kubescape/messaging/pulsar/config"
1313
pulsarconnector "github.com/kubescape/messaging/pulsar/connector"
1414
"github.com/kubescape/synchronizer/domain"
@@ -127,14 +127,12 @@ func LoadClusterConfig() (armometadata.ClusterConfig, error) {
127127
return *clusterConfig, err
128128
}
129129

130-
func LoadServiceURLs(filePath string) (schema.IBackendServices, error) {
131-
pathAndFileName, present := os.LookupEnv("SERVICES")
132-
if !present {
133-
pathAndFileName = filePath
130+
func LoadServiceURLs(apiURL string) (schema.IBackendServices, error) {
131+
client, err := v3.NewServiceDiscoveryClientV3(apiURL)
132+
if err != nil {
133+
return nil, err
134134
}
135-
return servicediscovery.GetServices(
136-
v2.NewServiceDiscoveryFileV2(pathAndFileName),
137-
)
135+
return servicediscovery.GetServices(client)
138136
}
139137

140138
func (c *InCluster) ValidateConfig() error {

‎config/config_test.go‎

Lines changed: 24 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,15 @@
11
package config
22

33
import (
4+
"encoding/json"
5+
"net/http"
6+
"net/http/httptest"
47
"os"
58
"testing"
69

710
"github.com/armosec/armoapi-go/armotypes"
811
"github.com/armosec/utils-k8s-go/armometadata"
9-
"github.com/kubescape/backend/pkg/servicediscovery/schema"
10-
v2 "github.com/kubescape/backend/pkg/servicediscovery/v2"
12+
v3 "github.com/kubescape/backend/pkg/servicediscovery/v3"
1113
pulsarconfig "github.com/kubescape/messaging/pulsar/config"
1214
"github.com/kubescape/synchronizer/domain"
1315
"github.com/stretchr/testify/assert"
@@ -121,48 +123,28 @@ func TestLoadConfig(t *testing.T) {
121123
}
122124

123125
func TestLoadServiceURLs(t *testing.T) {
124-
tests := []struct {
125-
name string
126-
env map[string]string
127-
filePath string
128-
want schema.IBackendServices
129-
}{
130-
{
131-
name: "via filePath",
132-
filePath: "../configuration/services.json",
133-
want: &v2.ServicesV2{
134-
EventReceiverHttpUrl: "https://er-test.com",
135-
EventReceiverWebsocketUrl: "wss://er-test.com",
136-
GatewayUrl: "https://gw.test.com",
137-
ApiServerUrl: "https://api.test.com",
138-
MetricsUrl: "https://metrics.test.com",
139-
SynchronizerUrl: "ws://127.0.0.1:8080",
126+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
127+
w.Header().Set("Content-Type", "application/json")
128+
_ = json.NewEncoder(w).Encode(map[string]interface{}{
129+
"version": "v3",
130+
"response": map[string]string{
131+
"event-receiver-http": "https://er-test.com",
132+
"api-server": "https://api.test.com",
133+
"metrics": "https://metrics.test.com",
134+
"synchronizer": "ws://127.0.0.1:8080",
140135
},
141-
},
142-
{
143-
name: "via env",
144-
env: map[string]string{"SERVICES": "../configuration/services.json"},
145-
want: &v2.ServicesV2{
146-
EventReceiverHttpUrl: "https://er-test.com",
147-
EventReceiverWebsocketUrl: "wss://er-test.com",
148-
GatewayUrl: "https://gw.test.com",
149-
ApiServerUrl: "https://api.test.com",
150-
MetricsUrl: "https://metrics.test.com",
151-
SynchronizerUrl: "ws://127.0.0.1:8080",
152-
},
153-
},
154-
}
155-
for _, tt := range tests {
156-
t.Run(tt.name, func(t *testing.T) {
157-
for k, v := range tt.env {
158-
err := os.Setenv(k, v)
159-
assert.NoError(t, err)
160-
}
161-
got, err := LoadServiceURLs(tt.filePath)
162-
assert.NoError(t, err)
163-
assert.Equal(t, tt.want, got)
164136
})
165-
}
137+
}))
138+
defer srv.Close()
139+
140+
got, err := LoadServiceURLs(srv.URL)
141+
assert.NoError(t, err)
142+
assert.Equal(t, &v3.ServicesV3{
143+
EventReceiverHttpUrl: "https://er-test.com",
144+
ApiServerUrl: "https://api.test.com",
145+
MetricsUrl: "https://metrics.test.com",
146+
SynchronizerUrl: "ws://127.0.0.1:8080",
147+
}, got)
166148
}
167149

168150
func TestResource_String(t *testing.T) {

0 commit comments

Comments
 (0)