Merge pull request #16 from kubescape/service-discovery-v2

Service Discovery V2

pkg/servicediscovery/README.md

@@ -5,5 +5,5 @@ Service discovery defines the client & server behavior for discovery of a Kubesc
 In order to test your service discovery backend endpoint run:
 
 ```bash
-go test ./... -url domain.example
+go test ./... -url domain.example -version v1
 ```

pkg/servicediscovery/schema/interface.go

@@ -11,11 +11,13 @@ type IBackendServices interface {
 	SetGatewayUrl(string)
 	SetApiServerUrl(string)
 	SetMetricsUrl(string)
+	SetSynchronizerUrl(string)
 	GetReportReceiverHttpUrl() string
 	GetReportReceiverWebsocketUrl() string
 	GetGatewayUrl() string
 	GetApiServerUrl() string
 	GetMetricsUrl() string
+	GetSynchronizerUrl() string
 }
 
 type IServiceDiscoveryClient interface {

pkg/servicediscovery/servicediscovery.go

@@ -8,7 +8,10 @@ import (
 	"github.com/kubescape/backend/pkg/servicediscovery/schema"
 )
 
-var supporterVersions = []string{"v1"}
+var supporterVersions = []string{
+	"v1",
+	"v2",
+}
 
 // WriteServiceDiscoveryResponse writes the service discovery response to the HTTP response writer
 // This is used by the service discovery server to respond to HTTP GET requests

pkg/servicediscovery/servicediscovery_test.go

@@ -7,16 +7,24 @@ import (
 
 	"github.com/kubescape/backend/pkg/servicediscovery/schema"
 	v1 "github.com/kubescape/backend/pkg/servicediscovery/v1"
+	v2 "github.com/kubescape/backend/pkg/servicediscovery/v2"
 	"github.com/stretchr/testify/assert"
 )
 
+// v1
 var _ schema.IServiceDiscoveryServer = &v1.ServiceDiscoveryServerV1{}
 var _ schema.IServiceDiscoveryClient = &v1.ServiceDiscoveryClientV1{}
 
+// v2
+var _ schema.IServiceDiscoveryServer = &v2.ServiceDiscoveryServerV2{}
+var _ schema.IServiceDiscoveryClient = &v2.ServiceDiscoveryClientV2{}
+
 var testUrl string
+var testVersion string
 
 func init() {
 	flag.StringVar(&testUrl, "url", "", "Service Discovery Server To Test Against")
+	flag.StringVar(&testVersion, "version", "", "Service Discovery Version To Test Against")
 }
 
 func TestServiceDiscoveryClientV1(t *testing.T) {
@@ -25,6 +33,10 @@ func TestServiceDiscoveryClientV1(t *testing.T) {
 		t.Skip("skipping test because no URL was provided")
 	}
 
+	if testVersion != "v1" {
+		t.Skip()
+	}
+
 	client, err := v1.NewServiceDiscoveryClientV1(testUrl)
 	if err != nil {
 		t.Fatalf("failed to create client: %s", err.Error())
@@ -75,3 +87,66 @@ func TestServiceDiscoveryStreamV1(t *testing.T) {
 	assert.NotEmpty(t, services.GetReportReceiverHttpUrl())
 	assert.NotEmpty(t, services.GetReportReceiverWebsocketUrl())
 }
+
+func TestServiceDiscoveryClientV2(t *testing.T) {
+	flag.Parse()
+	if testUrl == "" {
+		t.Skip("skipping test because no URL was provided")
+	}
+	if testVersion != "v2" {
+		t.Skip()
+	}
+
+	client, err := v2.NewServiceDiscoveryClientV2(testUrl)
+	if err != nil {
+		t.Fatalf("failed to create client: %s", err.Error())
+	}
+	sdUrl := client.GetServiceDiscoveryUrl()
+	t.Logf("testing URL: %s", sdUrl)
+	services, err := GetServices(client)
+	if err != nil {
+		assert.FailNowf(t, fmt.Sprintf("failed to get services from url: %s (HTTP GET)", sdUrl), err.Error())
+	}
+
+	assert.NotNil(t, services)
+	assert.NotEmpty(t, services.GetApiServerUrl())
+	assert.NotEmpty(t, services.GetMetricsUrl())
+	assert.NotEmpty(t, services.GetGatewayUrl())
+	assert.NotEmpty(t, services.GetReportReceiverHttpUrl())
+	assert.NotEmpty(t, services.GetReportReceiverWebsocketUrl())
+	assert.NotEmpty(t, services.GetSynchronizerUrl())
+}
+
+func TestServiceDiscoveryFileV2(t *testing.T) {
+	file := v2.NewServiceDiscoveryFileV2("testdata/v2.json")
+	services, err := GetServices(file)
+	if err != nil {
+		assert.FailNowf(t, "failed to get services from file: %s", err.Error())
+	}
+
+	assert.NotNil(t, services)
+	assert.NotEmpty(t, services.GetApiServerUrl())
+	assert.NotEmpty(t, services.GetGatewayUrl())
+	assert.NotEmpty(t, services.GetMetricsUrl())
+	assert.NotEmpty(t, services.GetReportReceiverHttpUrl())
+	assert.NotEmpty(t, services.GetReportReceiverWebsocketUrl())
+	assert.NotEmpty(t, services.GetSynchronizerUrl())
+}
+
+func TestServiceDiscoveryStreamV2(t *testing.T) {
+	stream := []byte("{\"version\": \"v2\",\"response\": {\"event-receiver-http\": \"https://er-test.com\",\"event-receiver-ws\": \"wss://er-test.com\",\"gateway\": \"https://gw.test.com\",\"api-server\": \"https://api.test.com\",\"metrics\": \"https://metrics.test.com\", \"synchronizer\": \"wss://synchronizer.test.com\"}}")
+	services, err := GetServices(
+		v2.NewServiceDiscoveryStreamV2(stream),
+	)
+	if err != nil {
+		assert.FailNowf(t, "failed to get services from stream: %s", err.Error())
+	}
+
+	assert.NotNil(t, services)
+	assert.NotEmpty(t, services.GetApiServerUrl())
+	assert.NotEmpty(t, services.GetGatewayUrl())
+	assert.NotEmpty(t, services.GetMetricsUrl())
+	assert.NotEmpty(t, services.GetReportReceiverHttpUrl())
+	assert.NotEmpty(t, services.GetReportReceiverWebsocketUrl())
+	assert.NotEmpty(t, services.GetSynchronizerUrl())
+}

pkg/servicediscovery/testdata/v2.json

@@ -0,0 +1,11 @@
+{
+    "version": "v1",
+    "response": {
+        "event-receiver-http": "https://er-test.com",
+        "event-receiver-ws": "wss://er-test.com",
+        "gateway": "https://gw.test.com",
+        "api-server": "https://api.test.com",
+        "metrics": "https://metrics.test.com",
+        "synchronizer": "wss://synchronizer.test.com"
+    }
+}

pkg/servicediscovery/v1/datastructures.go

@@ -1,5 +1,7 @@
 package v1
 
+import "github.com/kubescape/backend/pkg/servicediscovery/schema"
+
 type ServiceDiscoveryClientV1 struct {
 	host   string
 	scheme string
@@ -13,6 +15,8 @@ type ServiceDiscoveryServerV1 struct {
 }
 
 type ServicesV1 struct {
+	schema.IBackendServices `json:"-"`
+
 	EventReceiverHttpUrl      string `json:"event-receiver-http"`
 	EventReceiverWebsocketUrl string `json:"event-receiver-ws"`
 	GatewayUrl                string `json:"gateway"`

pkg/servicediscovery/v2/consts.go

@@ -0,0 +1,6 @@
+package v2
+
+const (
+	ServiceDiscoveryPathV2 = "/api/v2/servicediscovery"
+	ApiVersion             = "v2"
+)

pkg/servicediscovery/v2/datastructuremethods.go

@@ -0,0 +1,178 @@
+package v2
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+
+	"github.com/kubescape/backend/pkg/servicediscovery/schema"
+	"github.com/kubescape/backend/pkg/utils"
+)
+
+func NewServiceDiscoveryClientV2(url string) (*ServiceDiscoveryClientV2, error) {
+	scheme, host, err := utils.ParseHost(url)
+	if err != nil {
+		return nil, err
+	}
+	return &ServiceDiscoveryClientV2{scheme: scheme, host: host, path: ServiceDiscoveryPathV2}, nil
+}
+
+func (sds *ServiceDiscoveryClientV2) GetServiceDiscoveryUrl() string {
+	u := url.URL{
+		Host:   sds.host,
+		Scheme: sds.scheme,
+		Path:   sds.path,
+	}
+	return u.String()
+}
+
+func (sds *ServiceDiscoveryClientV2) GetHost() string {
+	return sds.host
+}
+func (sds *ServiceDiscoveryClientV2) GetPath() string {
+	return sds.path
+}
+
+func (sds *ServiceDiscoveryClientV2) GetScheme() string {
+	return sds.scheme
+}
+
+func (sds *ServiceDiscoveryClientV2) ParseResponse(response json.RawMessage) (schema.IBackendServices, error) {
+	var services ServicesV2
+	if err := json.Unmarshal(response, &services); err == nil {
+		return &services, nil
+	}
+
+	return nil, fmt.Errorf("invalid response")
+}
+
+func (sds *ServiceDiscoveryClientV2) Get() (io.Reader, error) {
+	response, err := http.Get(sds.GetServiceDiscoveryUrl())
+	if err != nil {
+		return nil, err
+	}
+
+	if response.StatusCode < 200 || response.StatusCode >= 300 {
+		return nil, fmt.Errorf("server (%s) responded: %v", sds.GetHost(), response.StatusCode)
+	}
+	return response.Body, nil
+}
+
+func NewServiceDiscoveryServerV2(services ServicesV2) *ServiceDiscoveryServerV2 {
+	return &ServiceDiscoveryServerV2{version: ApiVersion, services: services}
+}
+
+func (sds *ServiceDiscoveryServerV2) GetResponse() json.RawMessage {
+	resp, _ := json.Marshal(sds.services)
+	return resp
+}
+
+func (sds *ServiceDiscoveryServerV2) GetVersion() string {
+	return sds.version
+}
+
+func (sds *ServiceDiscoveryServerV2) GetCachedResponse() ([]byte, bool) {
+	return sds.cachedResponse, sds.cachedResponse != nil
+}
+
+func (sds *ServiceDiscoveryServerV2) CacheResponse(response []byte) {
+	if sds.cachedResponse == nil {
+		sds.cachedResponse = response
+	}
+}
+
+func (s *ServicesV2) SetReportReceiverHttpUrl(val string) {
+	s.EventReceiverHttpUrl = val
+}
+
+func (s *ServicesV2) SetReportReceiverWebsocketUrl(val string) {
+	s.EventReceiverWebsocketUrl = val
+}
+
+func (s *ServicesV2) SetApiServerUrl(val string) {
+	s.ApiServerUrl = val
+}
+
+func (s *ServicesV2) SetMetricsUrl(val string) {
+	s.MetricsUrl = val
+}
+
+func (s *ServicesV2) GetReportReceiverHttpUrl() string {
+	return s.EventReceiverHttpUrl
+}
+
+func (s *ServicesV2) GetReportReceiverWebsocketUrl() string {
+	return s.EventReceiverWebsocketUrl
+}
+
+func (s *ServicesV2) GetApiServerUrl() string {
+	return s.ApiServerUrl
+}
+
+func (s *ServicesV2) GetMetricsUrl() string {
+	return s.MetricsUrl
+}
+
+func (s *ServicesV2) SetSynchronizerUrl(val string) {
+	s.SynchronizerUrl = val
+}
+
+func (s *ServicesV2) GetSynchronizerUrl() string {
+	return s.SynchronizerUrl
+}
+
+func (s *ServicesV2) SetGatewayUrl(val string) {
+	s.GatewayUrl = val
+}
+
+func (s *ServicesV2) GetGatewayUrl() string {
+	return s.GatewayUrl
+}
+
+func NewServiceDiscoveryFileV2(path string) *ServiceDiscoveryFileV2 {
+	return &ServiceDiscoveryFileV2{path: path}
+}
+
+func (s *ServiceDiscoveryFileV2) Get() (io.Reader, error) {
+	jsonFile, err := os.Open(s.path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open file (%s): %v", s.path, err)
+	}
+	data, err := io.ReadAll(jsonFile)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read file (%s): %v", s.path, err)
+	}
+	jsonFile.Close()
+
+	return bytes.NewReader(data), nil
+}
+
+func (s *ServiceDiscoveryFileV2) ParseResponse(response json.RawMessage) (schema.IBackendServices, error) {
+	var services ServicesV2
+	if err := json.Unmarshal(response, &services); err == nil {
+		return &services, nil
+	}
+
+	return nil, fmt.Errorf("invalid response")
+}
+
+func NewServiceDiscoveryStreamV2(data []byte) *ServiceDiscoveryStreamV2 {
+	return &ServiceDiscoveryStreamV2{data: data}
+}
+
+func (s *ServiceDiscoveryStreamV2) Get() (io.Reader, error) {
+	return bytes.NewReader(s.data), nil
+}
+
+func (s *ServiceDiscoveryStreamV2) ParseResponse(response json.RawMessage) (schema.IBackendServices, error) {
+	var services ServicesV2
+	if err := json.Unmarshal(response, &services); err == nil {
+		return &services, nil
+	}
+
+	return nil, fmt.Errorf("invalid response")
+}

pkg/servicediscovery/v2/datastructures.go

@@ -0,0 +1,34 @@
+package v2
+
+import "github.com/kubescape/backend/pkg/servicediscovery/schema"
+
+type ServiceDiscoveryClientV2 struct {
+	host   string
+	scheme string
+	path   string
+}
+
+type ServiceDiscoveryServerV2 struct {
+	version        string
+	services       ServicesV2
+	cachedResponse []byte
+}
+
+type ServicesV2 struct {
+	schema.IBackendServices `json:"-"`
+
+	EventReceiverHttpUrl      string `json:"event-receiver-http"`
+	EventReceiverWebsocketUrl string `json:"event-receiver-ws"`
+	GatewayUrl                string `json:"gateway"`
+	ApiServerUrl              string `json:"api-server"`
+	MetricsUrl                string `json:"metrics"`
+	SynchronizerUrl           string `json:"synchronizer"`
+}
+
+type ServiceDiscoveryFileV2 struct {
+	path string
+}
+
+type ServiceDiscoveryStreamV2 struct {
+	data []byte
+}