buildkit/client/client_utils_test.go

package client

import (
	"context"
	"encoding/binary"
	"encoding/json"
	"io"
	"os"
	"path/filepath"
	"testing"
	"time"

	ctd "github.com/containerd/containerd/v2/client"
	"github.com/containerd/containerd/v2/core/content"
	"github.com/containerd/containerd/v2/core/content/proxy"
	"github.com/containerd/containerd/v2/core/snapshots"
	"github.com/containerd/containerd/v2/pkg/namespaces"
	controlapi "github.com/moby/buildkit/api/services/control"
	"github.com/moby/buildkit/client/llb"
	"github.com/moby/buildkit/session/filesync"
	"github.com/moby/buildkit/util/testutil"
	"github.com/moby/buildkit/util/testutil/integration"
	"github.com/moby/buildkit/util/testutil/workers"
	ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
	"github.com/pkg/errors"
	"github.com/stretchr/testify/require"
	fsutiltypes "github.com/tonistiigi/fsutil/types"
)

func checkAllReleasable(t *testing.T, c *Client, sb integration.Sandbox, checkContent bool) {
	cl, err := c.ControlClient().ListenBuildHistory(sb.Context(), &controlapi.BuildHistoryRequest{
		EarlyExit: true,
	})
	require.NoError(t, err)

	for {
		resp, err := cl.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		require.NoError(t, err)
		_, err = c.ControlClient().UpdateBuildHistory(sb.Context(), &controlapi.UpdateBuildHistoryRequest{
			Ref:    resp.Record.Ref,
			Delete: true,
		})
		require.NoError(t, err)
	}

	retries := 0
loop0:
	for {
		require.Greater(t, 20, retries)
		retries++
		du, err := c.DiskUsage(sb.Context())
		require.NoError(t, err)
		for _, d := range du {
			if d.InUse {
				time.Sleep(500 * time.Millisecond)
				continue loop0
			}
		}
		break
	}

	err = c.Prune(sb.Context(), nil, PruneAll)
	require.NoError(t, err)

	du, err := c.DiskUsage(sb.Context())
	require.NoError(t, err)
	require.Equal(t, 0, len(du))

	// examine contents of exported tars (requires containerd)
	cdAddress := sb.ContainerdAddress()
	if cdAddress == "" {
		if checkContent {
			if err := workers.HasFeatureCompat(t, sb, workers.FeatureContentCheck); err == nil {
				store := proxy.NewContentStore(c.ContentClient())
				count := 0
				err := store.Walk(sb.Context(), func(info content.Info) error {
					count++
					return nil
				})
				require.NoError(t, err)
				require.Equal(t, 0, count)
			}
		}
		t.Logf("checkAllReleasable: skipping check for exported tars in non-containerd test")
		return
	}

	// TODO: make public pull helper function so this can be checked for standalone as well

	client, err := newContainerd(cdAddress)
	require.NoError(t, err)
	defer client.Close()

	for _, ns := range []string{"buildkit", "buildkit_history"} {
		ctx := namespaces.WithNamespace(sb.Context(), ns)
		snapshotterName := sb.Snapshotter()
		snapshotService := client.SnapshotService(snapshotterName)

		checkLeases := func() {
			leases, err := client.LeasesService().List(ctx)
			require.NoError(t, err)
			count := 0
			for _, l := range leases {
				_, isTemp := l.Labels["buildkit/lease.temporary"]
				_, isExpire := l.Labels["containerd.io/gc.expire"]
				if isTemp && isExpire {
					continue
				}
				resources, err := client.LeasesService().ListResources(ctx, l)
				require.NoError(t, err)
				if len(resources) == 0 {
					continue
				}
				count++
				t.Logf("lease: %v", l)
				for _, r := range resources {
					t.Logf("lease resource: %v", r)
				}
			}
			require.Equal(t, 0, count)
		}

		if checkContent {
			images, err := client.ImageService().List(ctx)
			require.NoError(t, err)
			for _, img := range images {
				err := client.ImageService().Delete(ctx, img.Name)
				require.NoError(t, err)
			}
		}

		retries = 0
		for {
			count := 0
			err = snapshotService.Walk(ctx, func(context.Context, snapshots.Info) error {
				count++
				return nil
			})
			require.NoError(t, err)
			if count == 0 {
				break
			}
			require.Less(t, retries, 20)
			retries++
			time.Sleep(500 * time.Millisecond)
		}

		if !checkContent {
			checkLeases()
			return
		}

		retries = 0
		for {
			count := 0
			var infos []content.Info
			err = client.ContentStore().Walk(ctx, func(info content.Info) error {
				count++
				infos = append(infos, info)
				return nil
			})
			require.NoError(t, err)
			if count == 0 {
				break
			}
			if retries >= 50 {
				for _, info := range infos {
					t.Logf("content: %v %v %+v", info.Digest, info.Size, info.Labels)
					ra, err := client.ContentStore().ReaderAt(ctx, ocispecs.Descriptor{
						Digest: info.Digest,
						Size:   info.Size,
					})
					if err == nil {
						dt := make([]byte, 1024)
						n, err := ra.ReadAt(dt, 0)
						t.Logf("data: %+v %q", err, string(dt[:n]))
					}
				}
				require.FailNowf(t, "content still exists", "%+v", infos)
			}
			retries++
			time.Sleep(500 * time.Millisecond)
		}

		checkLeases()
	}
}

func ensureFile(t *testing.T, path string) {
	st, err := os.Stat(path)
	require.NoError(t, err, "expected file at %s", path)
	require.True(t, st.Mode().IsRegular())
}

func ensureFileContents(t *testing.T, path, expectedContents string) {
	contents, err := os.ReadFile(path)
	require.NoError(t, err)

	actualContents := string(contents)
	require.Equal(t, expectedContents, actualContents,
		"file contents mismatch for %s\nexpected: %q\nactual: %q", path, expectedContents, actualContents)
}

// ensurePruneAll tries to ensure Prune completes with retries.
// Current cache implementation defers release-related logic using goroutine so
// there can be situation where a build has finished but the following prune doesn't
// cleanup cache because some records still haven't been released.
// This function tries to ensure prune by retrying it.
func ensurePruneAll(t *testing.T, c *Client, sb integration.Sandbox) {
	for i := range 2 {
		require.NoError(t, c.Prune(sb.Context(), nil, PruneAll))
		for range 20 {
			du, err := c.DiskUsage(sb.Context())
			require.NoError(t, err)
			if len(du) == 0 {
				return
			}
			time.Sleep(500 * time.Millisecond)
		}
		t.Logf("retrying prune(%d)", i)
	}
	t.Fatalf("failed to ensure prune")
}

func fixedWriteCloser(wc io.WriteCloser) filesync.FileOutputFunc {
	return func(map[string]string) (io.WriteCloser, error) {
		return wc, nil
	}
}

func newContainerd(cdAddress string) (*ctd.Client, error) {
	return ctd.New(cdAddress, ctd.WithTimeout(60*time.Second))
}

func parseFSMetadata(t *testing.T, dt []byte) []fsutiltypes.Stat {
	var m []fsutiltypes.Stat
	for len(dt) > 0 {
		var s fsutiltypes.Stat
		n := binary.LittleEndian.Uint32(dt[:4])
		dt = dt[4:]
		err := s.Unmarshal(dt[:n])
		require.NoError(t, err)
		m = append(m, *s.CloneVT())
		dt = dt[n:]
	}
	return m
}

func readFileInImage(ctx context.Context, t *testing.T, c *Client, ref, path string) ([]byte, error) {
	def, err := llb.Image(ref).Marshal(ctx)
	if err != nil {
		return nil, err
	}
	destDir := t.TempDir()

	_, err = c.Solve(ctx, def, SolveOpt{
		Exports: []ExportEntry{
			{
				Type:      ExporterLocal,
				OutputDir: destDir,
			},
		},
	}, nil)
	if err != nil {
		return nil, err
	}
	return os.ReadFile(filepath.Join(destDir, filepath.Clean(path)))
}

type imageTimestamps struct {
	FromImage      []string // from img.Created and img.[]History.Created
	FromAnnotation string   // from index.Manifests[0].Annotations["org.opencontainers.image.created"]
}

func readImageTimestamps(dt []byte) (*imageTimestamps, error) {
	m, err := testutil.ReadTarToMap(dt, false)
	if err != nil {
		return nil, err
	}

	if _, ok := m["oci-layout"]; !ok {
		return nil, errors.Errorf("no oci-layout")
	}

	var index ocispecs.Index
	if err := json.Unmarshal(m[ocispecs.ImageIndexFile].Data, &index); err != nil {
		return nil, err
	}
	if len(index.Manifests) != 1 {
		return nil, errors.Errorf("invalid manifest count %d", len(index.Manifests))
	}

	var res imageTimestamps
	res.FromAnnotation = index.Manifests[0].Annotations[ocispecs.AnnotationCreated]

	var mfst ocispecs.Manifest
	if err := json.Unmarshal(m[ocispecs.ImageBlobsDir+"/sha256/"+index.Manifests[0].Digest.Hex()].Data, &mfst); err != nil {
		return nil, err
	}
	// don't unmarshal to image type so we get the original string value
	type history struct {
		Created string `json:"created"`
	}

	img := struct {
		History []history `json:"history"`
		Created string    `json:"created"`
	}{}

	if err := json.Unmarshal(m[ocispecs.ImageBlobsDir+"/sha256/"+mfst.Config.Digest.Hex()].Data, &img); err != nil {
		return nil, err
	}

	res.FromImage = []string{
		img.Created,
	}
	for _, h := range img.History {
		res.FromImage = append(res.FromImage, h.Created)
	}
	return &res, nil
}

func requiresLinux(t *testing.T) {
	integration.SkipOnPlatform(t, "!linux")
}