[feat]: [CI-12621]: Parse Buildx Logs to get cache metrics (using channels)

app.go

@@ -317,6 +317,11 @@ func Run() {
 			Usage:  "Artifact file location that will be generated by the plugin. This file will include information of docker images that are uploaded by the plugin.",
 			EnvVar: "PLUGIN_ARTIFACT_FILE",
 		},
+		cli.StringFlag{
+			Name:   "cache-metrics-file",
+			Usage:  "Location of cache metrics file that will be generated by the plugin.",
+			EnvVar: "PLUGIN_CACHE_METRICS_FILE",
+		},
 		cli.StringFlag{
 			Name:   "registry-type",
 			Usage:  "registry type",
@@ -351,9 +356,10 @@ func run(c *cli.Context) error {
 			Config:      c.String("docker.config"),
 			AccessToken: c.String("access-token"),
 		},
-		CardPath:     c.String("drone-card-path"),
-		MetadataFile: c.String("metadata-file"),
-		ArtifactFile: c.String("artifact-file"),
+		CardPath:         c.String("drone-card-path"),
+		MetadataFile:     c.String("metadata-file"),
+		ArtifactFile:     c.String("artifact-file"),
+		CacheMetricsFile: c.String("cache-metrics-file"),
 		Build: Build{
 			Remote:      c.String("remote.url"),
 			Name:        c.String("commit.sha"),

docker.go

@@ -3,10 +3,14 @@ package docker
 import (
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"regexp"
+	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/drone-plugins/drone-plugin-lib/drone"
@@ -82,15 +86,16 @@ type (
 
 	// Plugin defines the Docker plugin parameters.
 	Plugin struct {
-		Login        Login   // Docker login configuration
-		Build        Build   // Docker build configuration
-		Builder      Builder // Docker Buildx builder configuration
-		Daemon       Daemon  // Docker daemon configuration
-		Dryrun       bool    // Docker push is skipped
-		Cleanup      bool    // Docker purge is enabled
-		CardPath     string  // Card path to write file to
-		MetadataFile string  // Location to write the metadata file
-		ArtifactFile string  // Artifact path to write file to
+		Login            Login   // Docker login configuration
+		Build            Build   // Docker build configuration
+		Builder          Builder // Docker Buildx builder configuration
+		Daemon           Daemon  // Docker daemon configuration
+		Dryrun           bool    // Docker push is skipped
+		Cleanup          bool    // Docker purge is enabled
+		CardPath         string  // Card path to write file to
+		MetadataFile     string  // Location to write the metadata file
+		ArtifactFile     string  // Artifact path to write file to
+		CacheMetricsFile string  // Location to write the cache metrics file
 	}
 
 	Card []struct {
@@ -119,8 +124,51 @@ type (
 	TagStruct struct {
 		Tag string `json:"Tag"`
 	}
+
+	LayerStatus struct {
+		Status string
+		Time   float64 // Time in seconds; only set for DONE layers
+	}
+
+	CacheMetrics struct {
+		TotalLayers int                 `json:"total_layers"`
+		Done        int                 `json:"done"`
+		Cached      int                 `json:"cached"`
+		Errored     int                 `json:"errored"`
+		Cancelled   int                 `json:"cancelled"`
+		Layers      map[int]LayerStatus `json:"layers"`
+	}
+
+	tee struct {
+		w      io.Writer
+		status chan string
+	}
 )
 
+func (t *tee) Write(p []byte) (n int, err error) {
+	n, err = t.w.Write(p)
+	if err != nil {
+		return n, err
+	}
+	select {
+	case t.status <- string(p):
+		// Successfully sent to the channel
+	default:
+		// Drop the message if the channel is full to avoid blocking
+	}
+	return n, nil
+}
+
+func (t *tee) Close() error {
+	close(t.status)
+	return nil
+}
+
+func Tee(w io.Writer) (*tee, <-chan string) {
+	status := make(chan string, 100) // Buffered channel to reduce the risk of dropping messages
+	return &tee{w: w, status: status}, status
+}
+
 // Exec executes the plugin step
 func (p Plugin) Exec() error {
 
@@ -232,8 +280,42 @@ func (p Plugin) Exec() error {
 		cmd.Stdout = os.Stdout
 		cmd.Stderr = os.Stderr
 		trace(cmd)
+		var err error
+		if isCommandBuildxBuild(cmd.Args) && p.CacheMetricsFile != "" {
+			// Create a tee writer and get the channel
+			teeWriter, statusCh := Tee(os.Stdout)
+
+			var goroutineErr error
+
+			var wg sync.WaitGroup
+			wg.Add(1)
+			// Run the command in a goroutine
+			go func() {
+				defer teeWriter.Close()
+				defer wg.Done()
+
+				cmd.Stdout = teeWriter
+				cmd.Stderr = teeWriter
+				goroutineErr = cmd.Run()
+			}()
+
+			// Run the parseCacheMetrics function and handle errors
+			cacheMetrics, err := parseCacheMetrics(statusCh)
+			if err != nil {
+				fmt.Printf("Could not parse cache metrics: %s", err)
+			}
+			wg.Wait()
 
-		err := cmd.Run()
+			if goroutineErr != nil {
+				return goroutineErr
+			}
+
+			if err := writeCacheMetrics(cacheMetrics, p.CacheMetricsFile); err != nil {
+				return err
+			}
+		} else {
+			err = cmd.Run()
+		}
 		if err != nil && isCommandPrune(cmd.Args) {
 			fmt.Printf("Could not prune system containers. Ignoring...\n")
 		} else if err != nil && isCommandRmi(cmd.Args) {
@@ -281,6 +363,64 @@ func (p Plugin) Exec() error {
 	return nil
 }
 
+func parseCacheMetrics(ch <-chan string) (CacheMetrics, error) {
+	var cacheMetrics CacheMetrics
+	cacheMetrics.Layers = make(map[int]LayerStatus) // Initialize the map
+
+	re := regexp.MustCompile(`#(\d+) (DONE|CACHED|ERRORED|CANCELLED)(?: ([0-9.]+)s)?`)
+
+	for line := range ch {
+		matches := re.FindAllStringSubmatch(line, -1)
+		for _, match := range matches {
+			if len(match) > 2 {
+				layerIndex, _ := strconv.Atoi(match[1])
+				status := match[2]
+				layerStatus := LayerStatus{Status: status}
+
+				switch status {
+				case "DONE":
+					cacheMetrics.Done++
+					if len(match) == 4 && match[3] != "" {
+						if duration, err := strconv.ParseFloat(match[3], 64); err == nil {
+							layerStatus.Time = duration
+						}
+					}
+				case "CACHED":
+					cacheMetrics.Cached++
+				case "ERRORED":
+					cacheMetrics.Errored++
+				case "CANCELLED":
+					cacheMetrics.Cancelled++
+				}
+				cacheMetrics.Layers[layerIndex] = layerStatus
+			}
+		}
+	}
+
+	cacheMetrics.TotalLayers = cacheMetrics.Done + cacheMetrics.Cached + cacheMetrics.Errored + cacheMetrics.Cancelled
+
+	return cacheMetrics, nil
+}
+
+func writeCacheMetrics(data CacheMetrics, filename string) error {
+	b, err := json.MarshalIndent(data, "", "\t")
+	if err != nil {
+		return fmt.Errorf("failed with err %s to marshal output %+v", err, data)
+	}
+
+	dir := filepath.Dir(filename)
+	err = os.MkdirAll(dir, 0644)
+	if err != nil {
+		return fmt.Errorf("failed with err %s to create %s directory for cache metrics file", err, dir)
+	}
+
+	err = os.WriteFile(filename, b, 0644)
+	if err != nil {
+		return fmt.Errorf("failed to write cache metrics to cache metrics file %s", filename)
+	}
+	return nil
+}
+
 func getDigest(metadataFile string) (string, error) {
 	file, err := os.Open(metadataFile)
 	if err != nil {
@@ -587,6 +727,11 @@ func commandDaemon(daemon Daemon) *exec.Cmd {
 	return exec.Command(dockerdExe, args...)
 }
 
+// helper to check if args match "docker buildx build"
+func isCommandBuildxBuild(args []string) bool {
+	return len(args) > 3 && args[1] == "buildx" && args[2] == "build"
+}
+
 // helper to check if args match "docker prune"
 func isCommandPrune(args []string) bool {
 	return len(args) > 3 && args[2] == "prune"