Telemetry endpoint for receiving data implemented in REAR client, server and related CRDs (#6) (#134)

Author: andreacv98

apis/reservation/v1alpha1/contract_types.go

@@ -52,6 +52,9 @@ type ContractSpec struct {
 
 	// NetworkRequests contains the reference to the resource containing the network requests.
 	NetworkRequests string `json:"networkRequests,omitempty"`
+
+	// IngressTelemetryEndpoint is the endpoint where the ingress telemetry is sent by the provider
+	IngressTelemetryEndpoint *TelemetryServer `json:"ingressTelemetryEndpoint,omitempty"`
 }
 
 // ContractStatus defines the observed state of Contract.

apis/reservation/v1alpha1/reservation_types.go

@@ -20,6 +20,14 @@ import (
 	nodecorev1alpha1 "github.com/fluidos-project/node/apis/nodecore/v1alpha1"
 )
 
+// TelemetryServer defines the telemetry server configuration.
+type TelemetryServer struct {
+	// Endpoint is the endpoint where the telemetry is sent by the provider
+	Endpoint string `json:"endpoint"`
+	// Intents is the list of intents
+	Intents []string `json:"intents,omitempty"`
+}
+
 // ReservationSpec defines the desired state of Reservation.
 type ReservationSpec struct {
 
@@ -43,6 +51,9 @@ type ReservationSpec struct {
 
 	// PeeringCandidate is the reference to the PeeringCandidate of the Reservation
 	PeeringCandidate nodecorev1alpha1.GenericRef `json:"peeringCandidate,omitempty"`
+
+	// IngressTelemetryEndpoint is the endpoint where the ingress telemetry is sent by the provider
+	IngressTelemetryEndpoint *TelemetryServer `json:"ingressTelemetryEndpoint,omitempty"`
 }
 
 // ReservationStatus defines the observed state of Reservation.

apis/reservation/v1alpha1/zz_generated.deepcopy.go

@@ -281,6 +281,22 @@ spec:
                     - lastUpdateTime
                     type: object
                 type: object
+              ingressTelemetryEndpoint:
+                description: IngressTelemetryEndpoint is the endpoint where the ingress
+                  telemetry is sent by the provider
+                properties:
+                  endpoint:
+                    description: Endpoint is the endpoint where the telemetry is sent
+                      by the provider
+                    type: string
+                  intents:
+                    description: Intents is the list of intents
+                    items:
+                      type: string
+                    type: array
+                required:
+                - endpoint
+                type: object
               networkRequests:
                 description: NetworkRequests contains the reference to the resource
                   containing the network requests.

deployments/node/crds/reservation.fluidos.eu_contracts.yaml

@@ -119,6 +119,22 @@ spec:
                 - data
                 - type
                 type: object
+              ingressTelemetryEndpoint:
+                description: IngressTelemetryEndpoint is the endpoint where the ingress
+                  telemetry is sent by the provider
+                properties:
+                  endpoint:
+                    description: Endpoint is the endpoint where the telemetry is sent
+                      by the provider
+                    type: string
+                  intents:
+                    description: Intents is the list of intents
+                    items:
+                      type: string
+                    type: array
+                required:
+                - endpoint
+                type: object
               peeringCandidate:
                 description: PeeringCandidate is the reference to the PeeringCandidate
                   of the Reservation

deployments/node/crds/reservation.fluidos.eu_reservations.yaml

@@ -144,6 +144,15 @@ func (g *Gateway) PurchaseFlavor(ctx context.Context, transactionID string,
 		return nil, err
 	}
 
+	// Get Reservation from the transaction
+	reservation, err := getters.GetReservationByTransactionID(ctx, g.client, transactionID)
+	if err != nil {
+		return nil, err
+	}
+
+	// Parse the IngressTelemetryEndpoint
+	ite := parseutil.ParseTelemetryServer(reservation.Spec.IngressTelemetryEndpoint)
+
 	klog.Infof("Transaction %s for flavor %s", transactionID, transaction.FlavorID)
 
 	var liqoCredentials *models.LiqoCredentials
@@ -156,7 +165,8 @@ func (g *Gateway) PurchaseFlavor(ctx context.Context, transactionID string,
 	}
 
 	body := models.PurchaseRequest{
-		LiqoCredentials: liqoCredentials,
+		LiqoCredentials:          liqoCredentials,
+		IngressTelemetryEndpoint: ite,
 	}
 
 	selectorBytes, err := json.Marshal(body)

pkg/rear-controller/gateway/client.go

@@ -541,7 +541,13 @@ func (g *Gateway) purchaseFlavor(w http.ResponseWriter, r *http.Request) {
 	// Create a new contract
 	klog.Infof("Creating a new contract...")
 	// Forge the contract object
-	contract = *resourceforge.ForgeContract(flavorSold, &transaction, liqoCredentials, sellerLiqoCredentials.ClusterID)
+	contract = *resourceforge.ForgeContract(
+		flavorSold,
+		&transaction,
+		liqoCredentials,
+		sellerLiqoCredentials.ClusterID,
+		purchase.IngressTelemetryEndpoint,
+	)
 	err = g.client.Create(context.Background(), &contract)
 	if err != nil {
 		klog.Errorf("Error creating the Contract: %s", err)

pkg/rear-controller/gateway/provider.go

@@ -126,6 +126,25 @@ func GetLiqoCredentials(ctx context.Context, cl client.Client) (*nodecorev1alpha
 	}, nil
 }
 
+// GetReservationByTransactionID retrieves the Reservation with the given transaction ID.
+func GetReservationByTransactionID(ctx context.Context, c client.Client, transactionID string) (*reservationv1alpha1.Reservation, error) {
+	// Get all the Reservations
+	reservations := &reservationv1alpha1.ReservationList{}
+	if err := c.List(ctx, reservations, client.InNamespace(flags.FluidosNamespace)); err != nil {
+		klog.Errorf("Error getting the Reservations: %s", err)
+		return nil, err
+	}
+
+	for i := range reservations.Items {
+		reservation := reservations.Items[i]
+		if reservation.Status.TransactionID == transactionID {
+			return &reservation, nil
+		}
+	}
+
+	return nil, fmt.Errorf("reservation with transaction ID %s not found", transactionID)
+}
+
 // GetAllocationByClusterIDSpec retrieves the name of the allocation with the given in its Specs.
 func GetAllocationByClusterIDSpec(ctx context.Context, c client.Client, clusterID string) (nodecorev1alpha1.AllocationList, error) {
 	filteredAllocations := nodecorev1alpha1.AllocationList{}

pkg/utils/getters/getters.go

@@ -20,6 +20,9 @@ type PurchaseRequest struct {
 	// This field is optional and should be used only if the buyer is the Liqo peering target cluster, based on the Flavor you are going to purchase.
 	// This field may be dismissed in the future version of Liqo.
 	LiqoCredentials *LiqoCredentials `json:"liqoCredentials,omitempty"`
+	// IngressTelemetryEndpoint is the endpoint where the buyer wants to receive the telemetry data.
+	// This field is optional and should be used only if continuous telemetry is needed.
+	IngressTelemetryEndpoint *TelemetryServer `json:"ingressTelemetryEndpoint,omitempty"`
 }
 
 // ReserveRequest is the request model for reserving a Flavor.

pkg/utils/models/gateway.go

@@ -39,6 +39,12 @@ type Transaction struct {
 	ExpirationTime string         `json:"expirationTime"`
 }
 
+// TelemetryServer represents a TelemetryServer object with its characteristics.
+type TelemetryServer struct {
+	Endpoint string   `json:"endpoint"`
+	Intents  []string `json:"intents"`
+}
+
 // Contract represents a Contract object with its characteristics.
 type Contract struct {
 	ContractID               string            `json:"contractID"`
@@ -52,6 +58,7 @@ type Contract struct {
 	ExtraInformation         map[string]string `json:"extraInformation,omitempty"`
 	Configuration            *Configuration    `json:"configuration,omitempty"`
 	NetworkRequests          string            `json:"networkAuthorizations,omitempty"`
+	IngressTelemetryEndpoint *TelemetryServer  `json:"ingressTelemetryEndpoint,omitempty"`
 }
 
 // LiqoCredentials contains the credentials of a Liqo cluster to establish a peering.