add image scan status call (#175)

* update kvisor to support pending images

* fix

* fix

castai/client.go

@@ -38,7 +38,7 @@ const (
 
 type Client interface {
 	SendLogs(ctx context.Context, req *LogEvent) error
-	SendImagesResourcesChange(ctx context.Context, report *ImagesResourcesChange) error
+	UpdateImageStatus(ctx context.Context, report *UpdateImagesStatusRequest) error
 	SendCISReport(ctx context.Context, report *KubeBenchReport) error
 	SendDeltaReport(ctx context.Context, report *Delta) error
 	SendLinterChecks(ctx context.Context, checks []LinterCheck) error
@@ -154,7 +154,7 @@ func (c *client) SendLogs(ctx context.Context, req *LogEvent) error {
 	return nil
 }
 
-func (c *client) SendImagesResourcesChange(ctx context.Context, report *ImagesResourcesChange) error {
+func (c *client) UpdateImageStatus(ctx context.Context, report *UpdateImagesStatusRequest) error {
 	return c.sendReport(ctx, report, ReportTypeImagesResourcesChange)
 }
 

castai/imagemeta_types.go

@@ -26,7 +26,14 @@ type OsInfo struct {
 	*types.OS           `json:",inline"`
 }
 
-type ImagesResourcesChange struct {
+const (
+	ImageScanStatusPending ImageScanStatus = "pending"
+	ImageScanStatusError   ImageScanStatus = "error"
+)
+
+type ImageScanStatus string
+
+type UpdateImagesStatusRequest struct {
 	FullSnapshot bool    `json:"full_snapshot,omitempty"`
 	Images       []Image `json:"images"`
 }
@@ -35,6 +42,8 @@ type Image struct {
 	ID              string          `json:"id"`
 	Architecture    string          `json:"architecture"`
 	ResourcesChange ResourcesChange `json:"resourcesChange"`
+	Status          ImageScanStatus `json:"status,omitempty"`
+	ErrorMsg        string          `json:"errorMsg,omitempty"`
 }
 
 type ResourcesChange struct {

imagescan/controller.go

@@ -24,7 +24,7 @@ import (
 type castaiClient interface {
 	SendImageMetadata(ctx context.Context, meta *castai.ImageMetadata) error
 	GetSyncState(ctx context.Context, filter *castai.SyncStateFilter) (*castai.SyncStateResponse, error)
-	SendImagesResourcesChange(ctx context.Context, report *castai.ImagesResourcesChange) error
+	UpdateImageStatus(ctx context.Context, report *castai.UpdateImagesStatusRequest) error
 }
 
 func NewController(
@@ -151,15 +151,8 @@ func (s *Controller) handleDelta(event kube.Event, o kube.Object) {
 func (s *Controller) scheduleScans(ctx context.Context) (rerr error) {
 	s.syncFromRemoteState(ctx)
 
-	if s.fullSnapshotSent {
-		s.sendImagesResourcesChanges(ctx)
-	} else {
-		// Send initial full images current resources state once.
-		if err := s.sendFullSnapshotImageResources(ctx); err != nil {
-			s.log.Errorf("sending initial full images resources changes: %v", err)
-		} else {
-			s.fullSnapshotSent = true
-		}
+	if err := s.updateImageStatuses(ctx); err != nil {
+		s.log.Errorf("sending images resources changes: %v", err)
 	}
 
 	// Scan pending images.
@@ -182,21 +175,26 @@ func (s *Controller) scheduleScans(ctx context.Context) (rerr error) {
 	return nil
 }
 
+// Clear changes state for next scans
+func (s *Controller) clearOwnerState() {
+	for _, img := range s.delta.getImages() {
+		if img.ownerChanges.empty() {
+			continue
+		}
+		img.ownerChanges.clear()
+	}
+}
+
 func (s *Controller) findPendingImages() []*image {
 	images := s.delta.getImages()
 
 	now := s.timeGetter()
 
-	var privateImagesCount int
+	privateImagesCount := lo.CountBy(images, func(v *image) bool {
+		return isImagePrivate(v)
+	})
 	pendingImages := lo.Filter(images, func(v *image, _ int) bool {
-		isPrivateImage := errors.Is(v.lastScanErr, errPrivateImage)
-		if isPrivateImage {
-			privateImagesCount++
-		}
-		return !v.scanned &&
-			len(v.owners) > 0 &&
-			!isPrivateImage &&
-			(v.nextScan.IsZero() || v.nextScan.Before(now))
+		return isImagePending(v, now)
 	})
 	sort.Slice(pendingImages, func(i, j int) bool {
 		return pendingImages[i].failures < pendingImages[j].failures
@@ -233,6 +231,9 @@ func (s *Controller) scanImages(ctx context.Context, images []*image) error {
 			if err := s.scanImage(ctx, img); err != nil {
 				log.Errorf("image scan failed: %v", err)
 				s.delta.setImageScanError(img, err)
+				if err := s.updateImageStatusAsFailed(ctx, img, err); err != nil {
+					s.log.Errorf("sending images resources changes: %v", err)
+				}
 				return
 			}
 			log.Info("image scan finished")
@@ -339,59 +340,75 @@ func (s *Controller) concurrentScansNumber() int {
 	return int(s.cfg.MaxConcurrentScans)
 }
 
-func (s *Controller) sendImagesResourcesChanges(ctx context.Context) {
+func (s *Controller) updateImageStatuses(ctx context.Context) error {
 	images := s.delta.getImages()
+	if s.fullSnapshotSent {
+		// Filter only images that have owner changes.
+		images = lo.Filter(images, func(img *image, _ int) bool {
+			return !img.ownerChanges.empty()
+		})
+		s.clearOwnerState()
+	}
+
+	if len(images) == 0 {
+		return nil
+	}
+	now := s.timeGetter()
 	var imagesChanges []castai.Image
 	for _, img := range images {
-		if img.ownerChanges.empty() {
-			continue
+		changedResourceIds := lo.Uniq(img.ownerChanges.addedIDS)
+		if s.fullSnapshotSent {
+			changedResourceIds = lo.Keys(img.owners)
+		}
+		var updatedStatus castai.ImageScanStatus
+		if isImagePending(img, now) {
+			updatedStatus = castai.ImageScanStatusPending
 		}
 		imagesChanges = append(imagesChanges, castai.Image{
 			ID:           img.id,
 			Architecture: img.architecture,
 			ResourcesChange: castai.ResourcesChange{
-				ResourceIDs: lo.Uniq(img.ownerChanges.addedIDS),
+				ResourceIDs: changedResourceIds,
 			},
+			Status: updatedStatus,
 		})
 	}
-	if len(imagesChanges) == 0 {
-		return
-	}
 
 	s.log.Info("sending images resources changes")
-	report := &castai.ImagesResourcesChange{
-		Images: imagesChanges,
+	report := &castai.UpdateImagesStatusRequest{
+		FullSnapshot: s.fullSnapshotSent,
+		Images:       imagesChanges,
 	}
-	if err := s.client.SendImagesResourcesChange(ctx, report); err != nil {
-		s.log.Errorf("sending images resources changes: %v", err)
-		return
+	err := s.client.UpdateImageStatus(ctx, report)
+	if err != nil {
+		return err
 	}
+	s.fullSnapshotSent = true
+	return nil
+}
 
-	// Clear changes state.
-	for _, img := range images {
-		if img.ownerChanges.empty() {
-			continue
-		}
-		img.ownerChanges.clear()
+func (s *Controller) updateImageStatusAsFailed(ctx context.Context, image *image, scanJobError error) error {
+	if image == nil {
+		return errors.New("image is missing")
+	}
+	var errorMsg string
+	if scanJobError != nil {
+		errorMsg = scanJobError.Error()
 	}
-}
 
-func (s *Controller) sendFullSnapshotImageResources(ctx context.Context) error {
-	s.log.Info("sending initial full images resources changes")
-	images := s.delta.getImages()
-	report := &castai.ImagesResourcesChange{
-		FullSnapshot: true,
+	updatedImage := castai.Image{
+		ID:           image.id,
+		Architecture: image.architecture,
+		Status:       castai.ImageScanStatusError,
+		ErrorMsg:     errorMsg,
 	}
-	for _, img := range images {
-		report.Images = append(report.Images, castai.Image{
-			ID:           img.id,
-			Architecture: img.architecture,
-			ResourcesChange: castai.ResourcesChange{
-				ResourceIDs: lo.Keys(img.owners),
-			},
-		})
+
+	s.log.Info("sending image failed status")
+	report := &castai.UpdateImagesStatusRequest{
+		Images: []castai.Image{updatedImage},
 	}
-	return s.client.SendImagesResourcesChange(ctx, report)
+
+	return s.client.UpdateImageStatus(ctx, report)
 }
 
 func (s *Controller) syncFromRemoteState(ctx context.Context) {
@@ -433,3 +450,14 @@ func (s *Controller) syncFromRemoteState(ctx context.Context) {
 	}
 	s.log.Infof("images updated from remote state, full_resync=%v, scanned_images=%d", resp.Images.FullResourcesResyncRequired, len(resp.Images.ScannedImages))
 }
+
+func isImagePending(v *image, now time.Time) bool {
+	return !v.scanned &&
+		len(v.owners) > 0 &&
+		!isImagePrivate(v) &&
+		(v.nextScan.IsZero() || v.nextScan.Before(now))
+}
+
+func isImagePrivate(v *image) bool {
+	return errors.Is(v.lastScanErr, errPrivateImage)
+}