Use the common etcd client in etos-iut (#83)

Author: andmat900

cmd/iut/main.go

@@ -25,14 +25,14 @@ import (
 	"time"
 
 	config "github.com/eiffel-community/etos-api/internal/configs/iut"
+	"github.com/eiffel-community/etos-api/internal/database/etcd"
 	"github.com/eiffel-community/etos-api/internal/logging"
 	server "github.com/eiffel-community/etos-api/internal/server"
 	"github.com/eiffel-community/etos-api/pkg/application"
 	"github.com/eiffel-community/etos-api/pkg/iut/v1alpha1"
 	"github.com/sirupsen/logrus"
 	"github.com/snowzach/rotatefilehook"
 	"go.elastic.co/ecslogrus"
-	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 // main sets up logging and starts up the webserver.
@@ -62,17 +62,10 @@ func main() {
 		"user_log":    false,
 	})
 
-	// Database connection test
-	cli, err := clientv3.New(clientv3.Config{
-		Endpoints:   []string{cfg.DatabaseURI()},
-		DialTimeout: 5 * time.Second,
-	})
-	if err != nil {
-		log.WithError(err).Fatal("failed to create etcd connection")
-	}
-
+	iutEtcdTreePrefix := "/iut"
+	db := etcd.New(cfg, logger, iutEtcdTreePrefix)
 	log.Info("Loading v1alpha1 routes")
-	v1alpha1App := v1alpha1.New(cfg, log, ctx, cli)
+	v1alpha1App := v1alpha1.New(cfg, log, ctx, db)
 	defer v1alpha1App.Close()
 	router := application.New(v1alpha1App)
 

internal/database/database.go

@@ -22,6 +22,10 @@ import (
 	"github.com/google/uuid"
 )
 
+type Deleter interface {
+	Delete() error
+}
+
 // Opener is the common interface for database clients
 type Opener interface {
 	Open(context.Context, uuid.UUID) io.ReadWriter

internal/database/etcd/etcd.go

@@ -32,6 +32,7 @@ import (
 // TODO: refactor the client so that it does not store data it fetched.
 // However, without it implementing the database.Opener interface would be more complex (methods readByte, read).
 type Etcd struct {
+	database.Deleter
 	cfg        config.Config
 	client     *clientv3.Client
 	ID         uuid.UUID
@@ -74,6 +75,7 @@ func (etcd Etcd) Write(p []byte) (int, error) {
 		return 0, errors.New("please create a new etcd client using Open")
 	}
 	key := fmt.Sprintf("%s/%s", etcd.treePrefix, etcd.ID.String())
+
 	_, err := etcd.client.Put(etcd.ctx, key, string(p))
 	if err != nil {
 		return 0, err
@@ -91,35 +93,50 @@ func (etcd *Etcd) readByte() byte {
 // Read reads data from etcd and returns p bytes to user
 func (etcd *Etcd) Read(p []byte) (n int, err error) {
 	if etcd.ID == uuid.Nil {
-		err = errors.New("please create a new etcd client using NewWithID")
-		return n, err
+		return 0, errors.New("please create a new etcd client using NewWithID")
 	}
 
 	key := fmt.Sprintf("%s/%s", etcd.treePrefix, etcd.ID.String())
 
 	if !etcd.hasRead {
 		resp, err := etcd.client.Get(etcd.ctx, key)
 		if err != nil {
-			return n, err
+			return 0, err
 		}
 		if len(resp.Kvs) == 0 {
-			return n, io.EOF
+			return 0, io.EOF
 		}
 		etcd.data = resp.Kvs[0].Value
 		etcd.hasRead = true
 	}
 
 	if len(etcd.data) == 0 {
-		return n, io.EOF
+		return 0, io.EOF
 	}
-	if c := cap(p); c > 0 {
-		for n < c {
-			p[n] = etcd.readByte()
-			n++
-			if len(etcd.data) == 0 {
-				return n, io.EOF
-			}
-		}
+
+	// Copy as much data as possible to p
+	// The copy function copies the minimum of len(p) and len(etcd.data) bytes from etcd.data to p
+	// It returns the number of bytes copied, which is stored in n
+	n = copy(p, etcd.data)
+
+	// Update etcd.data to remove the portion of data that has already been copied to p
+	// etcd.data[n:] creates a new slice that starts from the n-th byte to the end of the original slice
+	// This effectively removes the first n bytes from etcd.data, ensuring that subsequent reads start from the correct position
+	etcd.data = etcd.data[n:]
+
+	if n == 0 {
+		return 0, io.EOF
 	}
+
 	return n, nil
 }
+
+// Delete deletes the current key from the database
+func (etcd Etcd) Delete() error {
+	key := fmt.Sprintf("%s/%s", etcd.treePrefix, etcd.ID.String())
+	_, err := etcd.client.Delete(etcd.ctx, key)
+	if err != nil {
+		return fmt.Errorf("Failed to delete key %s: %s", key, err.Error())
+	}
+	return nil
+}

pkg/iut/v1alpha1/v1alpha1.go

@@ -19,16 +19,17 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"runtime"
 	"sync"
 	"time"
 
 	eiffelevents "github.com/eiffel-community/eiffelevents-sdk-go"
 	config "github.com/eiffel-community/etos-api/internal/configs/iut"
+	"github.com/eiffel-community/etos-api/internal/database"
 	"github.com/eiffel-community/etos-api/pkg/application"
 	packageurl "github.com/package-url/packageurl-go"
-	clientv3 "go.etcd.io/etcd/client/v3"
 
 	"github.com/google/uuid"
 	"github.com/julienschmidt/httprouter"
@@ -38,14 +39,14 @@ import (
 type V1Alpha1Application struct {
 	logger   *logrus.Entry
 	cfg      config.Config
-	database *clientv3.Client
+	database database.Opener
 	wg       *sync.WaitGroup
 }
 
 type V1Alpha1Handler struct {
 	logger   *logrus.Entry
 	cfg      config.Config
-	database *clientv3.Client
+	database database.Opener
 	wg       *sync.WaitGroup
 }
 
@@ -71,11 +72,11 @@ func (a *V1Alpha1Application) Close() {
 }
 
 // New returns a new V1Alpha1Application object/struct
-func New(cfg config.Config, log *logrus.Entry, ctx context.Context, cli *clientv3.Client) application.Application {
+func New(cfg config.Config, log *logrus.Entry, ctx context.Context, db database.Opener) application.Application {
 	return &V1Alpha1Application{
 		logger:   log,
 		cfg:      cfg,
-		database: cli,
+		database: db,
 		wg:       &sync.WaitGroup{},
 	}
 }
@@ -120,25 +121,30 @@ type StatusRequest struct {
 	Id uuid.UUID `json:"id"`
 }
 
-type StopRequest struct {
-	Id uuid.UUID `json:"id"`
-}
-
 // Start creates a number of IUTs and stores them in the ETCD database returning a checkout ID.
 func (h V1Alpha1Handler) Start(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+	identifier, err := uuid.Parse(r.Header.Get("X-Etos-Id"))
+	logger := h.logger.WithField("identifier", identifier).WithContext(r.Context())
+	if err != nil {
+		RespondWithError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
 	checkOutID := uuid.New()
 
 	w.Header().Set("X-Content-Type-Options", "nosniff")
 	w.Header().Set("Content-Type", "application/json")
 
 	var startReq StartRequest
 	if err := json.NewDecoder(r.Body).Decode(&startReq); err != nil {
+		logger.Errorf("Failed to decode request body: %s", r.Body)
 		RespondWithError(w, http.StatusBadRequest, err.Error())
 		return
 	}
 	defer r.Body.Close()
 	purl, err := packageurl.FromString(startReq.ArtifactIdentity)
+
 	if err != nil {
+		logger.Errorf("Failed to create a purl struct from artifact identity: %s", startReq.ArtifactIdentity)
 		RespondWithError(w, http.StatusBadRequest, err.Error())
 		return
 	}
@@ -149,74 +155,86 @@ func (h V1Alpha1Handler) Start(w http.ResponseWriter, r *http.Request, ps httpro
 	}
 	iuts, err := json.Marshal(purls)
 	if err != nil {
+		logger.Errorf("Failed to marshal purls: %s", purls)
 		RespondWithError(w, http.StatusInternalServerError, err.Error())
 		return
 	}
-	_, err = h.database.Put(r.Context(), fmt.Sprintf("/iut/%s", checkOutID.String()), string(iuts))
+	client := h.database.Open(r.Context(), identifier)
+	_, err = client.Write([]byte(string(iuts)))
 	if err != nil {
+		logger.Errorf("Failed to write to database: %s", string(iuts))
 		RespondWithError(w, http.StatusInternalServerError, err.Error())
 		return
 	}
 	startResp := StartResponse{Id: checkOutID}
+	logger.Debugf("Start response: %s", startResp)
 	w.WriteHeader(http.StatusOK)
 	response, _ := json.Marshal(startResp)
 	_, _ = w.Write(response)
 }
 
 // Status creates a simple DONE Status response with IUTs.
 func (h V1Alpha1Handler) Status(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
-	identifier := r.Header.Get("X-Etos-Id")
+	identifier, err := uuid.Parse(r.Header.Get("X-Etos-Id"))
+	if err != nil {
+		RespondWithError(w, http.StatusInternalServerError, err.Error())
+	}
 	logger := h.logger.WithField("identifier", identifier).WithContext(r.Context())
 
 	id, err := uuid.Parse(r.URL.Query().Get("id"))
+	client := h.database.Open(r.Context(), identifier)
+
+	data, err := io.ReadAll(client)
 
-	key := fmt.Sprintf("/iut/%s", id)
-	dbResp, err := h.database.Get(r.Context(), key)
 	if err != nil {
-		logger.Errorf("Failed to look up status request id: %s", id)
-		RespondWithError(w, http.StatusInternalServerError, err.Error())
-		return
-	}
-	if len(dbResp.Kvs) == 0 {
-		err = fmt.Errorf("No key found: %s", key)
+		logger.Errorf("Failed to look up status request id: %s, %s", identifier, err.Error())
 		RespondWithError(w, http.StatusInternalServerError, err.Error())
 		return
 	}
 	statusResp := StatusResponse{
 		Id:     id,
 		Status: "DONE",
 	}
-	if err = json.Unmarshal(dbResp.Kvs[0].Value, &statusResp.Iuts); err != nil {
+	if err = json.Unmarshal(data, &statusResp.Iuts); err != nil {
+		logger.Errorf("Failed to unmarshal data: %s", data)
 		RespondWithError(w, http.StatusInternalServerError, err.Error())
 		return
 	}
 	response, err := json.Marshal(statusResp)
 	if err != nil {
+		logger.Errorf("Failed to marshal status response: %s", statusResp)
 		RespondWithError(w, http.StatusInternalServerError, err.Error())
 		return
 	}
+	logger.Debugf("Status response: %s", statusResp)
 	w.WriteHeader(http.StatusOK)
 	_, _ = w.Write(response)
+
 }
 
 // Stop deletes the given IUTs from the database and returns an empty response.
 func (h V1Alpha1Handler) Stop(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
-	identifier := r.Header.Get("X-Etos-Id")
+	identifier, err := uuid.Parse(r.Header.Get("X-Etos-Id"))
+	if err != nil {
+		RespondWithError(w, http.StatusInternalServerError, err.Error())
+	}
 	logger := h.logger.WithField("identifier", identifier).WithContext(r.Context())
 
-	var stopReq StopRequest
-	defer r.Body.Close()
-	if err := json.NewDecoder(r.Body).Decode(&stopReq); err != nil {
-		logger.Errorf("Bad delete request: %s", err.Error())
-		RespondWithError(w, http.StatusBadRequest, err.Error())
-		return
+	client := h.database.Open(r.Context(), identifier)
+	deleter, canDelete := client.(database.Deleter)
+	if !canDelete {
+		logger.Warning("The database does not support delete. Writing nil.")
+		_, err = client.Write(nil)
+	} else {
+		err = deleter.Delete()
 	}
-	_, err := h.database.Delete(r.Context(), fmt.Sprintf("/iut/%s", stopReq.Id))
+
 	if err != nil {
-		logger.Errorf("Etcd delete failed: %s", err.Error())
+		logger.Errorf("Database delete failed: %s", err.Error())
 		RespondWithError(w, http.StatusInternalServerError, err.Error())
 		return
 	}
+	logger.Debugf("Stop request succeeded")
 	w.WriteHeader(http.StatusNoContent)
 }