service configuration deployment done

Author: ponachte

.gitignore

@@ -4,4 +4,4 @@
 .idea
 logs/*
 temp
-cli-config.json
+cli-config*.json

config/kind.yaml config/local.yamlconfig/kind.yaml renamed to config/local.yaml

@@ -22,18 +22,24 @@ serviceConfigurations:
       orchestration:
         type: Deployment
         spec:
-          replicas: 1
-          selector:
-            matchLabels:
-              app: fetch
-          template:
-            metadata:
-              labels:
+          apiVersion: apps/v1
+          kind: Deployment
+          metadata:
+            name: fetch
+          spec:
+            replicas: 1
+            selector:
+              matchLabels:
                 app: fetch
-            spec:
-              containers:
-                - name: fetch
-                  image: fetch
-                  env:
-                    - name: GET_URL
-                      value: "$(URL)"
+            template:
+              metadata:
+                labels:
+                  app: fetch
+              spec:
+                containers:
+                  - name: fetch
+                    image: fetch
+                    imagePullPolicy: Never
+                    env:
+                      - name: GET_URL
+                        value: "$(URL)"

config/services/fetch.yaml

@@ -43,6 +43,7 @@ serviceConfigurations:
               containers:
                 - name: pacsoi
                   image: pacsoi-service
+                  imagePullPolicy: Never
                   env:
                     - name: SOURCES
                       value: "$(SOURCES)"

config/services/pacsoi.yaml

@@ -253,24 +253,16 @@ func (collec *ServiceCollection) postService(w http.ResponseWriter, r *http.Requ
 
 	for pred, output := range service.Configuration.Spec.OutputMapping {
 
-		if output.Distribution == nil || output.Distribution.Access == nil {
-			continue // abstract output → no endpoint
-		}
-
 		externalPath := output.Distribution.Access.ExternalPath
-
-		// Validate (now required)
 		if externalPath == "" {
 			errMsg := fmt.Sprintf("externalPath is required for output %s", pred)
 			logrus.Error(errMsg)
 			http.Error(w, errMsg, http.StatusInternalServerError)
 			return
 		}
 
-		// Normalize and join
 		path := util.JoinPaths(aggPath, externalPath)
 
-		// Prevent duplicates
 		if seen[path] {
 			errMsg := fmt.Sprintf("duplicate externalPath resolved to %s", path)
 			logrus.Error(errMsg)
@@ -279,15 +271,37 @@ func (collec *ServiceCollection) postService(w http.ResponseWriter, r *http.Requ
 		}
 		seen[path] = true
 
-		// Register handler
-		err = collec.HandleFunc(path, collec.HandleServiceOutput, []model.Scope{model.Read, model.Write})
+		// Build forward URL at registration time, nil if abstract
+		var forwardURL string
+		if output.Distribution != nil && output.Distribution.Access != nil {
+			access := output.Distribution.Access
+			internalPath := access.InternalPath
+			if internalPath == "" {
+				internalPath = "/"
+			}
+			if !strings.HasPrefix(internalPath, "/") {
+				internalPath = "/" + internalPath
+			}
+			forwardURL = fmt.Sprintf(
+				"%s://%s.%s.svc.cluster.local:%d%s",
+				access.Protocol,
+				service.NamespaceID,
+				model.Namespace,
+				access.ServicePort,
+				internalPath,
+			)
+		}
+
+		err = collec.HandleFunc(path, func(w http.ResponseWriter, r *http.Request) {
+			collec.HandleServiceOutput(w, r, forwardURL)
+		}, []model.Scope{model.Read, model.Write})
 		if err != nil {
 			logrus.WithError(err).Errorf("Error registering handler for output %s", pred)
 			http.Error(w, "Failed to create service from request", http.StatusInternalServerError)
 			return
 		}
 
-		logrus.Infof("Registered endpoint: %s → %s", path, pred)
+		logrus.Infof("Registered endpoint: %s → %s", pred, path)
 	}
 
 	// Return service information
@@ -328,99 +342,40 @@ func (collec *ServiceCollection) deleteService(w http.ResponseWriter, _ *http.Re
 }
 
 // Handles all incoming service requests <service path>/<output>
-func (collec *ServiceCollection) HandleServiceOutput(w http.ResponseWriter, r *http.Request) {
-	// Trim leading/trailing slashes and split path
-	parts := strings.Split(strings.Trim(r.URL.Path, "/"), "/")
-	if len(parts) < 2 {
-		http.Error(w, "Invalid path, must be /<service-path>/<output>", http.StatusBadRequest)
-		return
-	}
-
-	// Last segment is the output
-	pred := parts[len(parts)-1]
-
-	// Everything else is the service path
-	servicePath := parts[:len(parts)-1]
-	serviceID := strings.Join(servicePath, "-")
-	service, ok := collec.services[serviceID]
-	if !ok {
-		http.Error(w, "Service not found", http.StatusNotFound)
-		return
-	}
-
-	// Lookup output mapping from ServiceConfiguration
-	output, exists := service.Configuration.Spec.OutputMapping[pred]
-	if !exists {
-		logrus.Errorf("No output mapping found for %s", pred)
-		http.Error(w, fmt.Sprintf("Requested service has no output %s", pred), http.StatusNotFound)
-		return
-	}
-
-	if output.Distribution == nil || output.Distribution.Access == nil {
-		logrus.Errorf("Output %s has no distribution access defined", pred)
-		http.Error(w, fmt.Sprintf("Output %s is not accessible", pred), http.StatusNotFound)
+func (collec *ServiceCollection) HandleServiceOutput(w http.ResponseWriter, r *http.Request, forwardURL string) {
+	if forwardURL == "" {
+		http.Error(w, "Output has no distribution", http.StatusNotFound)
 		return
 	}
 
-	access := output.Distribution.Access
-
-	// Extract servicePort
-	port := access.ServicePort
-
-	// Normalize internalPath (default "/")
-	internalPath := access.InternalPath
-	if internalPath == "" {
-		internalPath = "/"
-	}
-
-	// ensure it starts with "/"
-	if !strings.HasPrefix(internalPath, "/") {
-		internalPath = "/" + internalPath
-	}
-
-	// Build forwarding URL
-	forwardURL := fmt.Sprintf(
-		"%s://%s.%s.svc.cluster.local:%d%s",
-		access.Protocol,
-		service.NamespaceID,
-		model.Namespace,
-		port,
-		internalPath,
-	)
-
 	req, err := http.NewRequest(r.Method, forwardURL, r.Body)
 	if err != nil {
 		logrus.WithError(err).Error("Failed to create forward request")
 		http.Error(w, "Failed to reach requested service", http.StatusInternalServerError)
 		return
 	}
 
-	// Copy headers from original request
 	for k, vv := range r.Header {
 		for _, v := range vv {
 			req.Header.Add(k, v)
 		}
 	}
 
-	// Send request using proxy client
 	resp, err := model.HttpClient.Do(req)
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Failed to forward request: %v", err), http.StatusBadGateway)
 		return
 	}
 	defer resp.Body.Close()
 
-	// Copy response headers
 	for k, vv := range resp.Header {
 		for _, v := range vv {
 			w.Header().Add(k, v)
 		}
 	}
 
-	// Write status code
 	w.WriteHeader(resp.StatusCode)
 
-	// Copy response body
 	if _, err := io.Copy(w, resp.Body); err != nil {
 		logrus.WithError(err).Error("Failed to copy response body")
 	}

containers/aggregator/config/service_collection.go

@@ -189,14 +189,15 @@ func (service *Service) InitDescription() error {
 		description.Service.AddQuad(quad)
 	}
 
-	// OUTPUTS (UPDATED)
+	// OUTPUTS
 	for pred, output := range service.Configuration.Spec.OutputMapping {
 
+		id := uuid.New().String()
 		outputStore := rdfgo.NewStore()
 		description.Outputs[pred] = outputStore
 
-		// ✅ Dataset node
-		datasetNode := rdfgo.NewNamedNode(service.FullPath + "#" + pred + "Dataset")
+		// Dataset node
+		datasetNode := rdfgo.NewNamedNode(service.FullPath + "#" + id + "Dataset")
 
 		quad, err := rdfgo.NewQuad(
 			datasetNode,
@@ -209,7 +210,7 @@ func (service *Service) InitDescription() error {
 		}
 		outputStore.AddQuad(quad)
 
-		// ✅ Service serves dataset
+		// Service serves dataset
 		quad, err = rdfgo.NewQuad(
 			svcNode,
 			Dcat("servesDataset"),
@@ -221,7 +222,7 @@ func (service *Service) InitDescription() error {
 		}
 		outputStore.AddQuad(quad)
 
-		// ✅ DATASET METADATA
+		// DATASET METADATA
 		if output.Dataset != nil {
 
 			if output.Dataset.Title != "" {
@@ -250,7 +251,7 @@ func (service *Service) InitDescription() error {
 				outputStore.AddQuad(quad)
 			}
 
-			// ✅ Extra RDF properties
+			// Extra RDF properties
 			for predURI, val := range output.Dataset.ExtraProperties {
 				quad, err = rdfgo.NewQuad(
 					datasetNode,
@@ -265,10 +266,10 @@ func (service *Service) InitDescription() error {
 			}
 		}
 
-		// ✅ DISTRIBUTION
+		// DISTRIBUTION
 		if output.Distribution != nil && output.Distribution.Access != nil {
 
-			distNode := rdfgo.NewNamedNode(service.FullPath + "#" + pred + "Distribution")
+			distNode := rdfgo.NewNamedNode(service.FullPath + "#" + id + "Distribution")
 
 			// rdf:type
 			quad, err = rdfgo.NewQuad(
@@ -294,7 +295,7 @@ func (service *Service) InitDescription() error {
 			}
 			outputStore.AddQuad(quad)
 
-			// ✅ Distribution metadata
+			// Distribution metadata
 			if output.Distribution.Title != "" {
 				quad, err = rdfgo.NewQuad(
 					distNode,
@@ -321,7 +322,7 @@ func (service *Service) InitDescription() error {
 				outputStore.AddQuad(quad)
 			}
 
-			// ✅ Access URL
+			// Access URL
 			access := output.Distribution.Access
 
 			externalPath := access.ExternalPath
@@ -347,7 +348,7 @@ func (service *Service) InitDescription() error {
 			}
 			outputStore.AddQuad(quad)
 
-			// ✅ Link to DataService
+			// Link to DataService
 			quad, err = rdfgo.NewQuad(
 				distNode,
 				Dcat("accessService"),
@@ -359,7 +360,7 @@ func (service *Service) InitDescription() error {
 			}
 			outputStore.AddQuad(quad)
 
-			// ✅ Distribution extra RDF
+			// Distribution extra RDF
 			for predURI, val := range output.Distribution.ExtraProperties {
 				quad, err = rdfgo.NewQuad(
 					distNode,

containers/aggregator/model/service-description.go

@@ -3,6 +3,7 @@ package services
 import (
 	"aggregator/model"
 	"context"
+	"encoding/json"
 	"fmt"
 	"strings"
 	"time"
@@ -73,12 +74,31 @@ func deploy(
 		return err
 	}
 
+	// Inject UMA Proxy envs
+	if useUMA {
+		err = injectUMAEnv(obj)
+		if err != nil {
+			return err
+		}
+	}
+
+	// Inject namespace
+	err = injectNamespace(obj, model.Namespace)
+	if err != nil {
+		return err
+	}
+
 	// Inject labels
 	err = injectLabels(obj, service)
 	if err != nil {
 		return err
 	}
 
+	// Log final spec
+	if finalBytes, err := json.MarshalIndent(obj, "", "  "); err == nil {
+		logrus.Debugf("Final deployment spec for service %s:\n%s", service.NamespaceID, string(finalBytes))
+	}
+
 	// Apply
 	err = applyObject(ctx, obj)
 	if err != nil {

containers/aggregator/services/deploy-service.go

@@ -162,6 +162,20 @@ func injectIntoPodSpec(podSpec *corev1.PodSpec, envVars []corev1.EnvVar) {
 	}
 }
 
+func injectNamespace(obj interface{}, namespace string) error {
+	switch r := obj.(type) {
+	case *appsv1.Deployment:
+		r.Namespace = namespace
+	case *batchv1.Job:
+		r.Namespace = namespace
+	case *batchv1.CronJob:
+		r.Namespace = namespace
+	default:
+		return fmt.Errorf("unsupported object type")
+	}
+	return nil
+}
+
 func injectLabels(obj interface{}, service *model.Service) error {
 
 	labels := map[string]string{

containers/aggregator/services/util.go

@@ -12,6 +12,7 @@
 # ------------------------
 
 CONFIG_DIR  ?= ./config
+CONFIG_FILE ?= local.yaml
 SERVICE_DIR := $(CONFIG_DIR)/services
 FNO_DIR     := $(CONFIG_DIR)/fno
 SERVICE_FILES := $(wildcard $(SERVICE_DIR)/*.yaml)
@@ -34,7 +35,7 @@ deploy:
 
 kind-deploy:
 	$(MAKE) configure-etc-hosts HOSTS="aggregator.local wsl.local"
-	$(MAKE) deploy CONFIG=kind.yaml
+	$(MAKE) deploy CONFIG=$(CONFIG_FILE)
 
 undeploy:
 	@echo "🧹 Stopping aggregator deployment..."