Merge pull request #289 from meshery/MUzairS15/feat/fingerprinting/2

add processing on discovery of resources

Makefile

@@ -56,7 +56,7 @@ build:
 
 .PHONY: run-check
 ## Runs local instance of Meshsync: can be used during local development
-run: check nats	
+run: nats	
 	go$(v) mod tidy; \
 	DEBUG=true GOPROXY=direct GOSUMDB=off go run main.go
 

internal/pipeline/handlers.go

@@ -81,7 +81,7 @@ func (ri *RegisterInformer) publishItem(obj *unstructured.Unstructured, evtype b
 	err := ri.broker.Publish(config.PublishTo, &broker.Message{
 		ObjectType: broker.MeshSync,
 		EventType:  evtype,
-		Object:     model.ParseList(*obj),
+		Object:     model.ParseList(*obj, evtype),
 	})
 	if err != nil {
 		ri.log.Error(ErrPublish(config.Name, err))

meshsync/handlers.go

@@ -162,7 +162,7 @@ func (h *Handler) listStoreObjects() []model.KubernetesResource {
 	}
 	parsedObjects := make([]model.KubernetesResource, 0)
 	for _, obj := range objects {
-		parsedObjects = append(parsedObjects, model.ParseList(*obj.(*unstructured.Unstructured)))
+		parsedObjects = append(parsedObjects, model.ParseList(*obj.(*unstructured.Unstructured), broker.Add))
 	}
 	return parsedObjects
 }

pkg/model/model.go

@@ -19,7 +19,7 @@ type KubernetesResource struct {
 	Status                 *KubernetesResourceStatus     `json:"status,omitempty" gorm:"foreignkey:ID;references:id;constraint:OnUpdate:CASCADE,OnDelete:SET NULL;"`
 	ClusterID              string                        `json:"cluster_id"`
 	PatternResource        *uuid.UUID                    `json:"pattern_resource"`
-	ComponentMetadata      map[string]interface{}        `json:"component_metadata" gorm:"type:bytes;serializer:json"` 
+	ComponentMetadata      map[string]interface{}        `json:"component_metadata" gorm:"type:bytes;serializer:json"`
 	// Secondary fields for configsmaps and secrets
 	Immutable  string `json:"immutable,omitempty"`
 	Data       string `json:"data,omitempty"`

pkg/model/model_converter.go

@@ -6,18 +6,19 @@ import (
 
 	"github.com/buger/jsonparser"
 	"github.com/google/uuid"
+	"github.com/layer5io/meshkit/broker"
 	"github.com/layer5io/meshkit/utils"
 	"github.com/layer5io/meshsync/internal/config"
 	iutils "github.com/layer5io/meshsync/pkg/utils"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 )
 
-func ParseList(object unstructured.Unstructured) KubernetesResource {
+func ParseList(object unstructured.Unstructured, eventType broker.EventType) KubernetesResource {
 	data, _ := object.MarshalJSON()
 	result := KubernetesResource{}
-
 	_ = utils.Unmarshal(string(data), &result)
 
+	processorInstance := GetProcessorInstance(result.Kind)
 	// ObjectMeta internal models
 	labels := make([]*KubernetesKeyValue, 0)
 	_ = jsonparser.ObjectEach(data, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) error {
@@ -88,6 +89,9 @@ func ParseList(object unstructured.Unstructured) KubernetesResource {
 	}
 
 	result.ClusterID = iutils.GetClusterID()
+	if processorInstance != nil {
+		_ = processorInstance.Process(data, &result, eventType)
+	}
 
 	return result
 }

pkg/model/preprocessor.go

@@ -0,0 +1,18 @@
+package model
+
+import (
+	"github.com/layer5io/meshkit/broker"
+)
+
+type ProcessFunc interface {
+	Process(obj []byte, k8sresource *KubernetesResource, evtype broker.EventType) error
+}
+
+func GetProcessorInstance(kind string) ProcessFunc {
+	switch kind {
+	case "Service":
+		return &K8SService{}
+	default:
+		return nil
+	}
+}

pkg/model/process.go

@@ -0,0 +1,78 @@
+package model
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"strings"
+
+	"github.com/layer5io/meshkit/broker"
+	"github.com/layer5io/meshkit/utils"
+	"github.com/layer5io/meshkit/utils/kubernetes"
+	v1 "k8s.io/api/core/v1"
+)
+
+type K8SService struct{}
+
+func (s *K8SService) Process(data []byte, k8sresource *KubernetesResource, evtype broker.EventType) error {
+	if evtype == broker.Delete {
+		return nil
+	}
+
+	k8sservice := &v1.Service{}
+
+	err := utils.Unmarshal(string(data), k8sservice)
+	if err != nil {
+		return err
+	}
+
+	urls := []string{}
+	endpoint, err := kubernetes.GetEndpoint(context.Background(), &kubernetes.ServiceOptions{}, k8sservice)
+	if err != nil {
+		return err
+	}
+
+	if endpoint != nil {
+		if endpoint.External != nil {
+			url, err := s.validateURL(endpoint.External.Address, endpoint.External.Port)
+			if err == nil {
+				urls = append(urls, url)
+			}
+		}
+		if endpoint.Internal != nil {
+			url, err := s.validateURL(endpoint.Internal.Address, endpoint.Internal.Port)
+			if err == nil {
+				urls = append(urls, url)
+			}
+		}
+	}
+
+	if k8sresource.ComponentMetadata == nil {
+		k8sresource.ComponentMetadata = make(map[string]interface{})
+	}
+	k8sresource.ComponentMetadata = map[string]interface{}{
+		"capabilities": map[string]interface{}{
+                        // indicates that this svc can be upgraded to "Meshery Connection".
+			"connection": true,
+			"urls":       urls,
+		},
+	}
+
+	return nil
+}
+
+func (s *K8SService) validateURL(address string, port int32) (serviceurl string, err error) {
+	protocol := "http"
+	if port == 443 {
+		protocol = "https"
+	}
+
+	// For some Cluster IP type svc the address is set as None
+	// Hence to prevent adding these IPs as URLs, below check is added.
+	if strings.Contains(strings.ToLower(address), "none") {
+		return serviceurl, kubernetes.ErrEndpointNotFound
+	}
+	serviceurl = fmt.Sprintf("%s://%s:%d", protocol, address, port)
+	_, err = url.ParseRequestURI(serviceurl)
+	return serviceurl, err
+}