fix: do not get item from an unsync backend

engine/cdn/cdn_item.go

@@ -386,6 +386,7 @@ func (s *Service) getRandomItemUnitIDByItemID(ctx context.Context, itemID string
 
 	itemUnits = s.Units.FilterItemUnitReaderByType(itemUnits)
 	itemUnits = s.Units.FilterItemUnitFromBuffer(itemUnits)
+	itemUnits = s.Units.FilterNotSyncBackend(itemUnits)
 
 	if len(itemUnits) == 0 {
 		return "", "", sdk.WithStack(fmt.Errorf("unable to find item units for item with id: %s", itemID))

engine/cdn/cdn_log_engine.go

@@ -3,11 +3,9 @@ package cdn
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"strings"
 	"time"
 
-	gocache "github.com/patrickmn/go-cache"
 	"github.com/rockbears/log"
 
 	"github.com/ovh/cds/engine/cache"
@@ -126,21 +124,12 @@ func (s *Service) dequeueMessages(ctx context.Context, jobLogsQueueKey string, q
 					hms = append(hms, hm)
 				}
 
-				// Send TO CDS API
-				if err := s.sendToCDS(ctx, hms); err != nil {
-					err = sdk.WrapError(err, "unable to send log to API")
+				// Send TO CDN Buffer
+				if err := s.sendToBufferWithRetry(ctx, hms); err != nil {
+					err = sdk.WrapError(err, "unable to send log into buffer")
 					ctx = sdk.ContextWithStacktrace(ctx, err)
 					log.Error(ctx, err.Error())
 				}
-
-				// Send TO CDN Buffer
-				if s.cdnEnabled(ctx, hms[0].Signature.ProjectKey) {
-					if err := s.sendToBufferWithRetry(ctx, hms); err != nil {
-						err = sdk.WrapError(err, "unable to send log into buffer")
-						ctx = sdk.ContextWithStacktrace(ctx, err)
-						log.Error(ctx, err.Error())
-					}
-				}
 				nbMessages += len(msgs)
 				t1 = time.Now()
 			}
@@ -191,23 +180,3 @@ func (s *Service) canDequeue(ctx context.Context, jobID string) (string, error)
 	}
 	return jobQueueKey, nil
 }
-
-// Check if storage on CDN is enabled
-func (s *Service) cdnEnabled(ctx context.Context, projectKey string) bool {
-	cacheKey := fmt.Sprintf("cdn-job-logs-enabled-project-%s", projectKey)
-	enabledI, has := runCache.Get(cacheKey)
-	if has {
-		return enabledI.(bool)
-	}
-
-	resp, err := s.Client.FeatureEnabled(sdk.FeatureCDNJobLogs, map[string]string{
-		"project_key": projectKey,
-	})
-	if err != nil {
-		log.Error(ctx, "unable to get job logs feature for project %s: %v", projectKey, err)
-		return false
-	}
-	enabled := !resp.Exists || resp.Enabled
-	runCache.Set(cacheKey, enabled, gocache.DefaultExpiration)
-	return enabled
-}

engine/cdn/cdn_log_store.go

@@ -2,7 +2,6 @@ package cdn
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	"github.com/rockbears/log"
@@ -14,63 +13,6 @@ import (
 	"github.com/ovh/cds/sdk/cdn"
 )
 
-func (s *Service) sendToCDS(ctx context.Context, msgs []handledMessage) error {
-	switch {
-	case msgs[0].Signature.Service != nil:
-		for _, msg := range msgs {
-			// Format line
-			msg.Msg.Full = buildMessage(msg)
-			if msg.Signature.Service != nil {
-				logs := sdk.ServiceLog{
-					ServiceRequirementName: msg.Signature.Service.RequirementName,
-					ServiceRequirementID:   msg.Signature.Service.RequirementID,
-					WorkflowNodeJobRunID:   msg.Signature.JobID,
-					WorkflowNodeRunID:      msg.Signature.NodeRunID,
-					Val:                    msg.Msg.Full,
-				}
-				if err := s.Client.QueueServiceLogs(ctx, []sdk.ServiceLog{logs}); err != nil {
-					return err
-				}
-			}
-		}
-		return nil
-	default:
-		// Aggregate messages by step
-		hms := make(map[string]handledMessage, len(msgs))
-		for _, msg := range msgs {
-			// Format line
-			msg.Msg.Full = buildMessage(msg)
-
-			k := fmt.Sprintf("%d-%d-%d", msg.Signature.JobID, msg.Signature.NodeRunID, msg.Signature.Worker.StepOrder)
-			// Aggregates lines in a single message
-			if _, ok := hms[k]; ok {
-				full := hms[k].Msg.Full
-				msg.Msg.Full = fmt.Sprintf("%s%s", full, msg.Msg.Full)
-				hms[k] = msg
-			} else {
-				hms[k] = msg
-			}
-		}
-
-		// Send logs to CDS API by step
-		for _, hm := range hms {
-			now := time.Now()
-			l := sdk.Log{
-				JobID:        hm.Signature.JobID,
-				NodeRunID:    hm.Signature.NodeRunID,
-				LastModified: &now,
-				StepOrder:    hm.Signature.Worker.StepOrder,
-				Val:          hm.Msg.Full,
-			}
-			if err := s.Client.QueueSendLogs(ctx, hm.Signature.JobID, l); err != nil {
-				return err
-			}
-		}
-	}
-
-	return nil
-}
-
 func (s *Service) sendToBufferWithRetry(ctx context.Context, hms []handledMessage) error {
 	if len(hms) == 0 {
 		return nil

engine/cdn/cdn_metrics.go

@@ -122,6 +122,9 @@ func (s *Service) ComputeMetrics(ctx context.Context) {
 
 			var storageStats []storage.Stat
 			for _, su := range s.Units.Storages {
+				if !su.CanSync() {
+					continue
+				}
 				storageStats = append(storageStats, s.countItemsForUnit(ctx, su)...)
 			}
 

engine/cdn/storage/types.go

@@ -286,6 +286,26 @@ func (x RunningStorageUnits) GetBuffer(bufferType sdk.CDNItemType) BufferUnit {
 	}
 }
 
+func (x *RunningStorageUnits) CanSync(unitID string) bool {
+	for _, unit := range x.Storages {
+		if unit.ID() == unitID {
+			return unit.CanSync()
+		}
+	}
+	return false
+}
+
+func (x *RunningStorageUnits) FilterNotSyncBackend(ius []sdk.CDNItemUnit) []sdk.CDNItemUnit {
+	itemsUnits := make([]sdk.CDNItemUnit, 0, len(ius))
+	for _, u := range ius {
+		if !x.CanSync(u.UnitID) {
+			continue
+		}
+		itemsUnits = append(itemsUnits, u)
+	}
+	return itemsUnits
+}
+
 func (x *RunningStorageUnits) FilterItemUnitFromBuffer(ius []sdk.CDNItemUnit) []sdk.CDNItemUnit {
 	itemsUnits := make([]sdk.CDNItemUnit, 0, len(ius))
 	for _, u := range ius {