dilim/buildkit
1
0
Fork 0
mirror of https://github.com/moby/buildkit.git synced 2026-08-07 08:10:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2236 from morlay/master

Browse Source 
refactor to use `util/bklog.G(ctx)` instead `logrus.` directly
This commit is contained in:
Tõnis Tiigi
2021-07-13 00:25:22 -07:00
committed by GitHub
parent 1018595476 18b49fd7dc
commit 06e8602caa
41 changed files with 226 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
cache/migrate_v2.go vendored
View File

@@ -13,9 +13,9 @@ import (
"github.com/containerd/containerd/snapshots"
"github.com/moby/buildkit/cache/metadata"
"github.com/moby/buildkit/snapshot"
"github.com/moby/buildkit/util/bklog"
digest "github.com/opencontainers/go-digest"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
func migrateChainID(si *metadata.StorageItem, all map[string]*metadata.StorageItem) (digest.Digest, digest.Digest, error) {
@@ -252,7 +252,7 @@ func MigrateV2(ctx context.Context, from, to string, cs content.Store, s snapsho
}
for _, item := range byID {
logrus.Infof("migrated %s parent:%q snapshot:%v committed:%v blob:%v diffid:%v chainID:%v blobChainID:%v",
bklog.G(ctx).Infof("migrated %s parent:%q snapshot:%v committed:%v blob:%v diffid:%v chainID:%v blobChainID:%v",
item.ID(), getParent(item), getSnapshotID(item), getCommitted(item), getBlob(item), getDiffID(item), getChainID(item), getBlobChainID(item))
}

2
cache/remotecache/gha/gha.go vendored
View File

@@ -297,7 +297,7 @@ func (ci *importer) Resolve(ctx context.Context, _ specs.Descriptor, id string,
if err != nil {
return nil, err
}
cms = append(cms, solver.NewCacheManager(id, keysStorage, resultStorage))
cms = append(cms, solver.NewCacheManager(ctx, id, keysStorage, resultStorage))
}
return solver.NewCombinedCacheManager(cms, nil), nil

4
cache/remotecache/import.go vendored
View File

@@ -94,7 +94,7 @@ func (ci *contentCacheImporter) Resolve(ctx context.Context, desc ocispec.Descri
if err != nil {
return nil, err
}
return solver.NewCacheManager(id, keysStorage, resultStorage), nil
return solver.NewCacheManager(ctx, id, keysStorage, resultStorage), nil
}
func readBlob(ctx context.Context, provider content.Provider, desc ocispec.Descriptor) ([]byte, error) {
@@ -229,7 +229,7 @@ func (ci *contentCacheImporter) importInlineCache(ctx context.Context, dt []byte
if err != nil {
return nil, err
}
cms = append(cms, solver.NewCacheManager(id, keysStorage, resultStorage))
cms = append(cms, solver.NewCacheManager(ctx, id, keysStorage, resultStorage))
}
return solver.NewCombinedCacheManager(cms, nil), nil

12
client/solve.go
View File

@@ -20,10 +20,10 @@ import (
"github.com/moby/buildkit/session/filesync"
"github.com/moby/buildkit/session/grpchijack"
"github.com/moby/buildkit/solver/pb"
"github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/util/entitlements"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
fstypes "github.com/tonistiigi/fsutil/types"
"go.opentelemetry.io/otel/trace"
"golang.org/x/sync/errgroup"
@@ -109,7 +109,7 @@ func (c *Client) solve(ctx context.Context, def *llb.Definition, runGateway runG
}
}
cacheOpt, err := parseCacheOptions(opt)
cacheOpt, err := parseCacheOptions(ctx, opt)
if err != nil {
return nil, err
}
@@ -181,7 +181,7 @@ func (c *Client) solve(ctx context.Context, def *llb.Definition, runGateway runG
<-time.After(3 * time.Second)
cancelStatus()
}()
logrus.Debugf("stopping session")
bklog.G(ctx).Debugf("stopping session")
s.Close()
}()
var pbd *pb.Definition
@@ -370,7 +370,7 @@ type cacheOptions struct {
frontendAttrs map[string]string
}
func parseCacheOptions(opt SolveOpt) (*cacheOptions, error) {
func parseCacheOptions(ctx context.Context, opt SolveOpt) (*cacheOptions, error) {
var (
cacheExports []*controlapi.CacheOptionsEntry
cacheImports []*controlapi.CacheOptionsEntry
@@ -423,14 +423,14 @@ func parseCacheOptions(opt SolveOpt) (*cacheOptions, error) {
}
cs, err := contentlocal.NewStore(csDir)
if err != nil {
logrus.Warning("local cache import at " + csDir + " not found due to err: " + err.Error())
bklog.G(ctx).Warning("local cache import at " + csDir + " not found due to err: " + err.Error())
continue
}
// if digest is not specified, load from "latest" tag
if attrs["digest"] == "" {
idx, err := ociindex.ReadIndexJSONFileLocked(filepath.Join(csDir, "index.json"))
if err != nil {
logrus.Warning("local cache import at " + csDir + " not found due to err: " + err.Error())
bklog.G(ctx).Warning("local cache import at " + csDir + " not found due to err: " + err.Error())
continue
}
for _, m := range idx.Manifests {

10
cmd/buildkitd/main.go
View File

@@ -16,6 +16,7 @@ import (
"time"
"github.com/BurntSushi/toml"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/pkg/seed"
"github.com/containerd/containerd/pkg/userns"
"github.com/containerd/containerd/platforms"
@@ -45,8 +46,8 @@ import (
"github.com/moby/buildkit/util/appcontext"
"github.com/moby/buildkit/util/appdefaults"
"github.com/moby/buildkit/util/archutil"
"github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/util/grpcerrors"
_ "github.com/moby/buildkit/util/log"
"github.com/moby/buildkit/util/profiler"
"github.com/moby/buildkit/util/resolver"
"github.com/moby/buildkit/util/stack"
@@ -76,6 +77,9 @@ func init() {
seed.WithTimeAndRand()
reexec.Init()
// overwrites containerd/log.G
log.G = bklog.GetLogger
}
var propagators = propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{})
@@ -296,10 +300,10 @@ func main() {
err = ctx.Err()
}
logrus.Infof("stopping server")
bklog.G(ctx).Infof("stopping server")
if os.Getenv("NOTIFY_SOCKET") != "" {
notified, notifyErr := sddaemon.SdNotify(false, sddaemon.SdNotifyStopping)
logrus.Debugf("SdNotifyStopping notified=%v, err=%v", notified, notifyErr)
bklog.G(ctx).Debugf("SdNotifyStopping notified=%v, err=%v", notified, notifyErr)
}
server.GracefulStop()

3
cmd/buildkitd/main_containerd_worker.go
View File

@@ -3,6 +3,7 @@
package main
import (
"context"
"os"
"strconv"
"strings"
@@ -249,7 +250,7 @@ func containerdWorkerInitializer(c *cli.Context, common workerInitializerOpt) ([
}
opt.Platforms = platforms
}
w, err := base.NewWorker(opt)
w, err := base.NewWorker(context.TODO(), opt)
if err != nil {
return nil, err
}

2
cmd/buildkitd/main_oci_worker.go
View File

@@ -301,7 +301,7 @@ func ociWorkerInitializer(c *cli.Context, common workerInitializerOpt) ([]worker
}
opt.Platforms = platforms
}
w, err := base.NewWorker(opt)
w, err := base.NewWorker(context.TODO(), opt)
if err != nil {
return nil, err
}

33
control/control.go
View File

@@ -6,6 +6,8 @@ import (
"sync/atomic"
"time"
"github.com/moby/buildkit/util/bklog"
controlapi "github.com/moby/buildkit/api/services/control"
apitypes "github.com/moby/buildkit/api/types"
"github.com/moby/buildkit/cache/remotecache"
@@ -23,7 +25,6 @@ import (
"github.com/moby/buildkit/util/tracing/transform"
"github.com/moby/buildkit/worker"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
tracev1 "go.opentelemetry.io/proto/otlp/collector/trace/v1"
v1 "go.opentelemetry.io/proto/otlp/collector/trace/v1"
@@ -57,7 +58,7 @@ type Controller struct { // TODO: ControlService
}
func NewController(opt Opt) (*Controller, error) {
cache := solver.NewCacheManager("local", opt.CacheKeyStorage, worker.NewCacheResultStorage(opt.WorkerController))
cache := solver.NewCacheManager(context.TODO(), "local", opt.CacheKeyStorage, worker.NewCacheResultStorage(opt.WorkerController))
gatewayForwarder := controlgateway.NewGatewayForwarder()
@@ -142,7 +143,7 @@ func (c *Controller) Prune(req *controlapi.PruneRequest, stream controlapi.Contr
ReleaseUnreferenced() error
}); ok {
if err := c.ReleaseUnreferenced(); err != nil {
logrus.Errorf("failed to release cache metadata: %+v", err)
bklog.G(ctx).Errorf("failed to release cache metadata: %+v", err)
}
}
}
@@ -284,7 +285,11 @@ func (c *Controller) Solve(ctx context.Context, req *controlapi.SolveRequest) (*
if err != nil {
return nil, err
}
cacheExportMode = parseCacheExportMode(e.Attrs["mode"])
if exportMode, supported := parseCacheExportMode(e.Attrs["mode"]); !supported {
bklog.G(ctx).Debugf("skipping invalid cache export mode: %s", e.Attrs["mode"])
} else {
cacheExportMode = exportMode
}
}
for _, im := range req.Cache.Imports {
cacheImports = append(cacheImports, frontend.CacheOptionsEntry{
@@ -384,7 +389,8 @@ func (c *Controller) Status(req *controlapi.StatusRequest, stream controlapi.Con
}
func (c *Controller) Session(stream controlapi.Control_SessionServer) error {
logrus.Debugf("session started")
bklog.G(stream.Context()).Debugf("session started")
conn, closeCh, opts := grpchijack.Hijack(stream)
defer conn.Close()
@@ -395,7 +401,7 @@ func (c *Controller) Session(stream controlapi.Control_SessionServer) error {
}()
err := c.opt.SessionManager.HandleConn(ctx, conn, opts)
logrus.Debugf("session finished: %v", err)
bklog.G(ctx).Debugf("session finished: %v", err)
return err
}
@@ -451,25 +457,22 @@ func (c *Controller) gc() {
err = eg.Wait()
close(ch)
if err != nil {
logrus.Errorf("gc error: %+v", err)
bklog.G(ctx).Errorf("gc error: %+v", err)
}
<-done
if size > 0 {
logrus.Debugf("gc cleaned up %d bytes", size)
bklog.G(ctx).Debugf("gc cleaned up %d bytes", size)
}
}
func parseCacheExportMode(mode string) solver.CacheExportMode {
func parseCacheExportMode(mode string) (solver.CacheExportMode, bool) {
switch mode {
case "min":
return solver.CacheExportModeMin
return solver.CacheExportModeMin, true
case "max":
return solver.CacheExportModeMax
case "":
default:
logrus.Debugf("skipping invalid cache export mode: %s", mode)
return solver.CacheExportModeMax, true
}
return solver.CacheExportModeMin
return solver.CacheExportModeMin, false
}
func toPBGCPolicy(in []client.PruneInfo) []*apitypes.GCPolicy {

7
executor/containerdexecutor/executor.go
View File

@@ -11,6 +11,8 @@ import (
"syscall"
"time"
"github.com/moby/buildkit/util/bklog"
"github.com/containerd/containerd"
"github.com/containerd/containerd/cio"
"github.com/containerd/containerd/mount"
@@ -26,7 +28,6 @@ import (
"github.com/moby/buildkit/util/network"
"github.com/opencontainers/runtime-spec/specs-go"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
type containerdExecutor struct {
@@ -153,7 +154,7 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
defer namespace.Close()
if meta.NetMode == pb.NetMode_HOST {
logrus.Info("enabling HostNetworking")
bklog.G(ctx).Info("enabling HostNetworking")
}
opts := []containerdoci.SpecOpts{oci.WithUIDGID(uid, gid, sgids)}
@@ -364,7 +365,7 @@ func (w *containerdExecutor) runProcess(ctx context.Context, p containerd.Proces
}
err = p.Resize(resizeCtx, size.Cols, size.Rows)
if err != nil {
logrus.Warnf("Failed to resize %s: %s", p.ID(), err)
bklog.G(resizeCtx).Warnf("Failed to resize %s: %s", p.ID(), err)
}
}
}

9
executor/runcexecutor/executor.go
View File

@@ -12,6 +12,8 @@ import (
"syscall"
"time"
"github.com/moby/buildkit/util/bklog"
"github.com/containerd/containerd/mount"
containerdoci "github.com/containerd/containerd/oci"
"github.com/containerd/continuity/fs"
@@ -27,7 +29,6 @@ import (
"github.com/moby/buildkit/util/stack"
"github.com/opencontainers/runtime-spec/specs-go"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
type Opt struct {
@@ -166,7 +167,7 @@ func (w *runcExecutor) Run(ctx context.Context, id string, root executor.Mount,
defer namespace.Close()
if meta.NetMode == pb.NetMode_HOST {
logrus.Info("enabling HostNetworking")
bklog.G(ctx).Info("enabling HostNetworking")
}
resolvConf, err := oci.GetResolvConf(ctx, w.root, w.idmap, w.dns)
@@ -303,7 +304,7 @@ func (w *runcExecutor) Run(ctx context.Context, id string, root executor.Mount,
case <-ctx.Done():
killCtx, timeout := context.WithTimeout(context.Background(), 7*time.Second)
if err := w.runc.Kill(killCtx, id, int(syscall.SIGKILL), nil); err != nil {
logrus.Errorf("failed to kill runc %s: %+v", id, err)
bklog.G(ctx).Errorf("failed to kill runc %s: %+v", id, err)
select {
case <-killCtx.Done():
timeout()
@@ -324,7 +325,7 @@ func (w *runcExecutor) Run(ctx context.Context, id string, root executor.Mount,
}
}()
logrus.Debugf("> creating %s %v", id, meta.Args)
bklog.G(ctx).Debugf("> creating %s %v", id, meta.Args)
// this is a cheat, we have not actually started, but as close as we can get with runc for now
if started != nil {
startedOnce.Do(func() {

8
executor/runcexecutor/executor_linux.go
View File

@@ -11,9 +11,9 @@ import (
runc "github.com/containerd/go-runc"
"github.com/docker/docker/pkg/signal"
"github.com/moby/buildkit/executor"
"github.com/moby/buildkit/util/bklog"
"github.com/opencontainers/runtime-spec/specs-go"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/sync/errgroup"
)
@@ -74,7 +74,7 @@ func (w *runcExecutor) callWithIO(ctx context.Context, id, bundle string, proces
cancel() // this will shutdown resize loop
err := eg.Wait()
if err != nil {
logrus.Warningf("error while shutting down tty io: %s", err)
bklog.G(ctx).Warningf("error while shutting down tty io: %s", err)
}
}()
@@ -135,11 +135,11 @@ func (w *runcExecutor) callWithIO(ctx context.Context, id, bundle string, proces
Width: uint16(resize.Cols),
})
if err != nil {
logrus.Errorf("failed to resize ptm: %s", err)
bklog.G(ctx).Errorf("failed to resize ptm: %s", err)
}
err = runcProcess.Signal(signal.SIGWINCH)
if err != nil {
logrus.Errorf("failed to send SIGWINCH to process: %s", err)
bklog.G(ctx).Errorf("failed to send SIGWINCH to process: %s", err)
}
}
}

10
exporter/containerimage/writer.go
View File

@@ -8,6 +8,8 @@ import (
"strings"
"time"
"github.com/moby/buildkit/util/bklog"
"github.com/containerd/containerd/content"
"github.com/containerd/containerd/diff"
"github.com/containerd/containerd/images"
@@ -25,7 +27,6 @@ import (
specs "github.com/opencontainers/image-spec/specs-go"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/sync/errgroup"
)
@@ -197,7 +198,7 @@ func (ic *ImageWriter) commitDistributionManifest(ctx context.Context, ref cache
return nil, nil, err
}
remote, history = normalizeLayersAndHistory(remote, history, ref, oci)
remote, history = normalizeLayersAndHistory(ctx, remote, history, ref, oci)
config, err = patchImageConfig(config, remote.Descriptors, history, inlineCache)
if err != nil {
@@ -388,8 +389,7 @@ func patchImageConfig(dt []byte, descs []ocispec.Descriptor, history []ocispec.H
return dt, errors.Wrap(err, "failed to marshal config after patch")
}
func normalizeLayersAndHistory(remote *solver.Remote, history []ocispec.History, ref cache.ImmutableRef, oci bool) (*solver.Remote, []ocispec.History) {
func normalizeLayersAndHistory(ctx context.Context, remote *solver.Remote, history []ocispec.History, ref cache.ImmutableRef, oci bool) (*solver.Remote, []ocispec.History) {
refMeta := getRefMetadata(ref, len(remote.Descriptors))
var historyLayers int
@@ -402,7 +402,7 @@ func normalizeLayersAndHistory(remote *solver.Remote, history []ocispec.History,
if historyLayers > len(remote.Descriptors) {
// this case shouldn't happen but if it does force set history layers empty
// from the bottom
logrus.Warn("invalid image config with unaccounted layers")
bklog.G(ctx).Warn("invalid image config with unaccounted layers")
historyCopy := make([]ocispec.History, 0, len(history))
var l int
for _, h := range history {

7
frontend/gateway/container.go
View File

@@ -9,6 +9,8 @@ import (
"strings"
"sync"
"github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/cache"
"github.com/moby/buildkit/executor"
"github.com/moby/buildkit/frontend/gateway/client"
@@ -20,7 +22,6 @@ import (
utilsystem "github.com/moby/buildkit/util/system"
"github.com/moby/buildkit/worker"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/sync/errgroup"
)
@@ -331,7 +332,7 @@ func (gwCtr *gatewayContainer) Start(ctx context.Context, req client.StartReques
if !started {
startedCh := make(chan struct{})
gwProc.errGroup.Go(func() error {
logrus.Debugf("Starting new container for %s with args: %q", gwCtr.id, procInfo.Meta.Args)
bklog.G(gwCtr.ctx).Debugf("Starting new container for %s with args: %q", gwCtr.id, procInfo.Meta.Args)
err := gwCtr.executor.Run(ctx, gwCtr.id, gwCtr.rootFS, gwCtr.mounts, procInfo, startedCh)
return stack.Enable(err)
})
@@ -341,7 +342,7 @@ func (gwCtr *gatewayContainer) Start(ctx context.Context, req client.StartReques
}
} else {
gwProc.errGroup.Go(func() error {
logrus.Debugf("Execing into container %s with args: %q", gwCtr.id, procInfo.Meta.Args)
bklog.G(gwCtr.ctx).Debugf("Execing into container %s with args: %q", gwCtr.id, procInfo.Meta.Args)
err := gwCtr.executor.Exec(ctx, gwCtr.id, procInfo)
return stack.Enable(err)
})

27
frontend/gateway/gateway.go
View File

@@ -12,6 +12,8 @@ import (
"sync"
"time"
"github.com/moby/buildkit/util/bklog"
"github.com/docker/distribution/reference"
"github.com/gogo/googleapis/google/rpc"
gogotypes "github.com/gogo/protobuf/types"
@@ -41,7 +43,6 @@ import (
"github.com/opencontainers/go-digest"
specs "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/http2"
"golang.org/x/sync/errgroup"
spb "google.golang.org/genproto/googleapis/rpc/status"
@@ -834,7 +835,7 @@ func (lbf *llbBridgeForwarder) Inputs(ctx context.Context, in *pb.InputsRequest)
}
func (lbf *llbBridgeForwarder) NewContainer(ctx context.Context, in *pb.NewContainerRequest) (_ *pb.NewContainerResponse, err error) {
logrus.Debugf("|<--- NewContainer %s", in.ContainerID)
bklog.G(ctx).Debugf("|<--- NewContainer %s", in.ContainerID)
ctrReq := NewContainerRequest{
ContainerID: in.ContainerID,
NetMode: in.Network,
@@ -909,7 +910,7 @@ func (lbf *llbBridgeForwarder) NewContainer(ctx context.Context, in *pb.NewConta
}
func (lbf *llbBridgeForwarder) ReleaseContainer(ctx context.Context, in *pb.ReleaseContainerRequest) (*pb.ReleaseContainerResponse, error) {
logrus.Debugf("|<--- ReleaseContainer %s", in.ContainerID)
bklog.G(ctx).Debugf("|<--- ReleaseContainer %s", in.ContainerID)
lbf.ctrsMu.Lock()
ctr, ok := lbf.ctrs[in.ContainerID]
delete(lbf.ctrs, in.ContainerID)
@@ -1024,7 +1025,7 @@ type outputWriter struct {
}
func (w *outputWriter) Write(msg []byte) (int, error) {
logrus.Debugf("|---> File Message %s, fd=%d, %d bytes", w.processID, w.fd, len(msg))
bklog.G(w.stream.Context()).Debugf("|---> File Message %s, fd=%d, %d bytes", w.processID, w.fd, len(msg))
err := w.stream.Send(&pb.ExecMessage{
ProcessID: w.processID,
Input: &pb.ExecMessage_File{
@@ -1054,15 +1055,15 @@ func (lbf *llbBridgeForwarder) ExecProcess(srv pb.LLBBridge_ExecProcessServer) e
}
switch m := execMsg.GetInput().(type) {
case *pb.ExecMessage_Init:
logrus.Debugf("|<--- Init Message %s", execMsg.ProcessID)
bklog.G(ctx).Debugf("|<--- Init Message %s", execMsg.ProcessID)
case *pb.ExecMessage_File:
if m.File.EOF {
logrus.Debugf("|<--- File Message %s, fd=%d, EOF", execMsg.ProcessID, m.File.Fd)
bklog.G(ctx).Debugf("|<--- File Message %s, fd=%d, EOF", execMsg.ProcessID, m.File.Fd)
} else {
logrus.Debugf("|<--- File Message %s, fd=%d, %d bytes", execMsg.ProcessID, m.File.Fd, len(m.File.Data))
bklog.G(ctx).Debugf("|<--- File Message %s, fd=%d, %d bytes", execMsg.ProcessID, m.File.Fd, len(m.File.Data))
}
case *pb.ExecMessage_Resize:
logrus.Debugf("|<--- Resize Message %s", execMsg.ProcessID)
bklog.G(ctx).Debugf("|<--- Resize Message %s", execMsg.ProcessID)
}
select {
case <-ctx.Done():
@@ -1150,7 +1151,7 @@ func (lbf *llbBridgeForwarder) ExecProcess(srv pb.LLBBridge_ExecProcessServer) e
eg.Go(func() error {
<-pio.done
logrus.Debugf("|---> Done Message %s", pid)
bklog.G(ctx).Debugf("|---> Done Message %s", pid)
err := srv.Send(&pb.ExecMessage{
ProcessID: pid,
Input: &pb.ExecMessage_Done{
@@ -1182,7 +1183,7 @@ func (lbf *llbBridgeForwarder) ExecProcess(srv pb.LLBBridge_ExecProcessServer) e
if errors.As(err, &exitError) {
statusCode = exitError.ExitCode
}
logrus.Debugf("|---> Exit Message %s, code=%d, error=%s", pid, statusCode, err)
bklog.G(ctx).Debugf("|---> Exit Message %s, code=%d, error=%s", pid, statusCode, err)
sendErr := srv.Send(&pb.ExecMessage{
ProcessID: pid,
Input: &pb.ExecMessage_Exit{
@@ -1207,7 +1208,7 @@ func (lbf *llbBridgeForwarder) ExecProcess(srv pb.LLBBridge_ExecProcessServer) e
return stack.Enable(err)
})
logrus.Debugf("|---> Started Message %s", pid)
bklog.G(ctx).Debugf("|---> Started Message %s", pid)
err = srv.Send(&pb.ExecMessage{
ProcessID: pid,
Input: &pb.ExecMessage_Started{
@@ -1246,7 +1247,7 @@ func (lbf *llbBridgeForwarder) ExecProcess(srv pb.LLBBridge_ExecProcessServer) e
return stack.Enable(err)
}
// no error so must be EOF
logrus.Debugf("|---> File Message %s, fd=%d, EOF", pid, fd)
bklog.G(ctx).Debugf("|---> File Message %s, fd=%d, EOF", pid, fd)
err = srv.Send(&pb.ExecMessage{
ProcessID: pid,
Input: &pb.ExecMessage_File{
@@ -1288,7 +1289,7 @@ func serve(ctx context.Context, grpcServer *grpc.Server, conn net.Conn) {
<-ctx.Done()
conn.Close()
}()
logrus.Debugf("serving grpc connection")
bklog.G(ctx).Debugf("serving grpc connection")
(&http2.Server{}).ServeConn(conn, &http2.ServeConnOpts{Handler: grpcServer})
}

13
frontend/gateway/grpcclient/client.go
View File

@@ -11,6 +11,8 @@ import (
"sync"
"time"
"github.com/moby/buildkit/util/bklog"
"github.com/gogo/googleapis/google/rpc"
gogotypes "github.com/gogo/protobuf/types"
"github.com/golang/protobuf/ptypes/any"
@@ -24,7 +26,6 @@ import (
"github.com/moby/buildkit/util/grpcerrors"
digest "github.com/opencontainers/go-digest"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
fstypes "github.com/tonistiigi/fsutil/types"
"golang.org/x/sync/errgroup"
spb "google.golang.org/genproto/googleapis/rpc/status"
@@ -576,7 +577,7 @@ func (m *messageForwarder) Start() (err error) {
if errors.Is(err, io.EOF) || grpcerrors.Code(err) == codes.Canceled {
return nil
}
logrus.Debugf("|<--- %s", debugMessage(msg))
bklog.G(m.ctx).Debugf("|<--- %s", debugMessage(msg))
if err != nil {
return err
@@ -587,7 +588,7 @@ func (m *messageForwarder) Start() (err error) {
m.mu.Unlock()
if !ok {
logrus.Debugf("Received exec message for unregistered process: %s", msg.String())
bklog.G(m.ctx).Debugf("Received exec message for unregistered process: %s", msg.String())
continue
}
msgs.Send(m.ctx, msg)
@@ -625,7 +626,7 @@ func (m *messageForwarder) Send(msg *pb.ExecMessage) error {
if !ok {
return errors.Errorf("process %s has ended, not sending message %#v", msg.ProcessID, msg.Input)
}
logrus.Debugf("|---> %s", debugMessage(msg))
bklog.G(m.ctx).Debugf("|---> %s", debugMessage(msg))
return m.stream.Send(msg)
}
@@ -703,7 +704,7 @@ func (c *grpcClient) NewContainer(ctx context.Context, req client.NewContainerRe
})
}
logrus.Debugf("|---> NewContainer %s", id)
bklog.G(ctx).Debugf("|---> NewContainer %s", id)
_, err = c.client.NewContainer(ctx, &pb.NewContainerRequest{
ContainerID: id,
Mounts: mounts,
@@ -897,7 +898,7 @@ func (ctr *container) Start(ctx context.Context, req client.StartRequest) (clien
}
func (ctr *container) Release(ctx context.Context) error {
logrus.Debugf("|---> ReleaseContainer %s", ctr.id)
bklog.G(ctx).Debugf("|---> ReleaseContainer %s", ctr.id)
_, err := ctr.client.ReleaseContainer(ctx, &pb.ReleaseContainerRequest{
ContainerID: ctr.id,
})

5
session/filesync/diffcopy.go
View File

@@ -7,8 +7,9 @@ import (
"os"
"time"
"github.com/moby/buildkit/util/bklog"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/tonistiigi/fsutil"
fstypes "github.com/tonistiigi/fsutil/types"
"google.golang.org/grpc"
@@ -73,7 +74,7 @@ func (wc *streamWriterCloser) Close() error {
func recvDiffCopy(ds grpc.ClientStream, dest string, cu CacheUpdater, progress progressCb, differ fsutil.DiffType, filter func(string, *fstypes.Stat) bool) (err error) {
st := time.Now()
defer func() {
logrus.Debugf("diffcopy took: %v", time.Since(st))
bklog.G(ds.Context()).Debugf("diffcopy took: %v", time.Since(st))
}()
var cf fsutil.ChangeFunc
var ch fsutil.ContentHasher

4
session/grpc.go
View File

@@ -7,9 +7,9 @@ import (
"time"
grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
"github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/util/grpcerrors"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
"go.opentelemetry.io/otel/trace"
"golang.org/x/net/http2"
@@ -22,7 +22,7 @@ func serve(ctx context.Context, grpcServer *grpc.Server, conn net.Conn) {
<-ctx.Done()
conn.Close()
}()
logrus.Debugf("serving grpc connection")
bklog.G(ctx).Debugf("serving grpc connection")
(&http2.Server{}).ServeConn(conn, &http2.ServeConnOpts{Handler: grpcServer})
}

4
session/testutil/testutil.go
View File

@@ -6,7 +6,7 @@ import (
"net"
"time"
"github.com/sirupsen/logrus"
"github.com/moby/buildkit/util/bklog"
)
// Handler is function called to handle incoming connection
@@ -22,7 +22,7 @@ func TestStream(handler Handler) Dialer {
go func() {
err := handler(context.TODO(), s1, meta)
if err != nil {
logrus.Error(err)
bklog.G(ctx).Error(err)
}
s1.Close()
}()

8
solver/cache_test.go
View File

@@ -239,7 +239,7 @@ func TestInMemoryCacheSelectorNested(t *testing.T) {
func TestInMemoryCacheReleaseParent(t *testing.T) {
storage := NewInMemoryCacheStorage()
results := NewInMemoryResultStorage()
m := NewCacheManager(identity.NewID(), storage, results)
m := NewCacheManager(context.TODO(), identity.NewID(), storage, results)
res0 := testResult("result0")
cacheFoo, err := m.Save(NewCacheKey(dgst("foo"), 0), res0, time.Now())
@@ -291,7 +291,7 @@ func TestInMemoryCacheReleaseParent(t *testing.T) {
func TestInMemoryCacheRestoreOfflineDeletion(t *testing.T) {
storage := NewInMemoryCacheStorage()
results := NewInMemoryResultStorage()
m := NewCacheManager(identity.NewID(), storage, results)
m := NewCacheManager(context.TODO(), identity.NewID(), storage, results)
res0 := testResult("result0")
cacheFoo, err := m.Save(NewCacheKey(dgst("foo"), 0), res0, time.Now())
@@ -305,7 +305,7 @@ func TestInMemoryCacheRestoreOfflineDeletion(t *testing.T) {
_, err = results2.Save(res1, time.Now()) // only add bar
require.NoError(t, err)
m = NewCacheManager(identity.NewID(), storage, results2)
m = NewCacheManager(context.TODO(), identity.NewID(), storage, results2)
keys, err := m.Query(nil, 0, dgst("foo"), 0)
require.NoError(t, err)
@@ -327,7 +327,7 @@ func TestInMemoryCacheRestoreOfflineDeletion(t *testing.T) {
func TestCarryOverFromSublink(t *testing.T) {
storage := NewInMemoryCacheStorage()
results := NewInMemoryResultStorage()
m := NewCacheManager(identity.NewID(), storage, results)
m := NewCacheManager(context.TODO(), identity.NewID(), storage, results)
cacheFoo, err := m.Save(NewCacheKey(dgst("foo"), 0), testResult("resultFoo"), time.Now())
require.NoError(t, err)

8
solver/cachemanager.go
View File

@@ -8,17 +8,17 @@ import (
"time"
"github.com/moby/buildkit/identity"
"github.com/moby/buildkit/util/bklog"
digest "github.com/opencontainers/go-digest"
"github.com/sirupsen/logrus"
)
// NewInMemoryCacheManager creates a new in-memory cache manager
func NewInMemoryCacheManager() CacheManager {
return NewCacheManager(identity.NewID(), NewInMemoryCacheStorage(), NewInMemoryResultStorage())
return NewCacheManager(context.TODO(), identity.NewID(), NewInMemoryCacheStorage(), NewInMemoryResultStorage())
}
// NewCacheManager creates a new cache manager with specific storage backend
func NewCacheManager(id string, storage CacheKeyStorage, results CacheResultStorage) CacheManager {
func NewCacheManager(ctx context.Context, id string, storage CacheKeyStorage, results CacheResultStorage) CacheManager {
cm := &cacheManager{
id: id,
backend: storage,
@@ -26,7 +26,7 @@ func NewCacheManager(id string, storage CacheKeyStorage, results CacheResultStor
}
if err := cm.ReleaseUnreferenced(); err != nil {
logrus.Errorf("failed to release unreferenced cache metadata: %+v", err)
bklog.G(ctx).Errorf("failed to release unreferenced cache metadata: %+v", err)
}
return cm

9
solver/cacheopts.go
View File

@@ -3,8 +3,9 @@ package solver
import (
"context"
"github.com/moby/buildkit/util/bklog"
digest "github.com/opencontainers/go-digest"
"github.com/sirupsen/logrus"
)
type CacheOpts map[interface{}]interface{}
@@ -27,7 +28,7 @@ func withAncestorCacheOpts(ctx context.Context, start *state) context.Context {
keySet[k] = struct{}{}
}
values := make(map[interface{}]interface{})
walkAncestors(start, func(st *state) bool {
walkAncestors(ctx, start, func(st *state) bool {
if st.clientVertex.Error != "" {
// don't use values from cancelled or otherwise error'd vertexes
return false
@@ -52,7 +53,7 @@ func withAncestorCacheOpts(ctx context.Context, start *state) context.Context {
})
}
func walkAncestors(start *state, f func(*state) bool) {
func walkAncestors(ctx context.Context, start *state, f func(*state) bool) {
stack := [][]*state{{start}}
cache := make(map[digest.Digest]struct{})
for len(stack) > 0 {
@@ -79,7 +80,7 @@ func walkAncestors(start *state, f func(*state) bool) {
parent := st.solver.actives[parentDgst]
st.solver.mu.RUnlock()
if parent == nil {
logrus.Warnf("parent %q not found in active job list during cache opt search", parentDgst)
bklog.G(ctx).Warnf("parent %q not found in active job list during cache opt search", parentDgst)
continue
}
stack[len(stack)-1] = append(stack[len(stack)-1], parent)

22
solver/edge.go
View File

@@ -6,9 +6,9 @@ import (
"time"
"github.com/moby/buildkit/solver/internal/pipe"
"github.com/moby/buildkit/util/bklog"
digest "github.com/opencontainers/go-digest"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
type edgeStatusType int
@@ -171,7 +171,7 @@ func (e *edge) finishIncoming(req pipe.Sender) {
err = context.Canceled
}
if debugScheduler {
logrus.Debugf("finishIncoming %s %v %#v desired=%s", e.edge.Vertex.Name(), err, e.edgeState, req.Request().Payload.(*edgeRequest).desiredState)
bklog.G(context.TODO()).Debugf("finishIncoming %s %v %#v desired=%s", e.edge.Vertex.Name(), err, e.edgeState, req.Request().Payload.(*edgeRequest).desiredState)
}
req.Finalize(&e.edgeState, err)
}
@@ -179,7 +179,7 @@ func (e *edge) finishIncoming(req pipe.Sender) {
// updateIncoming updates the current value of incoming pipe request
func (e *edge) updateIncoming(req pipe.Sender) {
if debugScheduler {
logrus.Debugf("updateIncoming %s %#v desired=%s", e.edge.Vertex.Name(), e.edgeState, req.Request().Payload.(*edgeRequest).desiredState)
bklog.G(context.TODO()).Debugf("updateIncoming %s %#v desired=%s", e.edge.Vertex.Name(), e.edgeState, req.Request().Payload.(*edgeRequest).desiredState)
}
req.Update(&e.edgeState)
}
@@ -358,7 +358,7 @@ func (e *edge) unpark(incoming []pipe.Sender, updates, allPipes []pipe.Receiver,
if e.execReq == nil {
if added := e.createInputRequests(desiredState, f, false); !added && !e.hasActiveOutgoing && !cacheMapReq {
logrus.Errorf("buildkit scheluding error: leaving incoming open. forcing solve. Please report this with BUILDKIT_SCHEDULER_DEBUG=1")
bklog.G(context.TODO()).Errorf("buildkit scheluding error: leaving incoming open. forcing solve. Please report this with BUILDKIT_SCHEDULER_DEBUG=1")
debugSchedulerPreUnpark(e, incoming, updates, allPipes)
e.createInputRequests(desiredState, f, true)
}
@@ -397,12 +397,12 @@ func (e *edge) processUpdate(upt pipe.Receiver) (depChanged bool) {
if !e.op.IgnoreCache() {
keys, err := e.op.Cache().Query(nil, 0, e.cacheMap.Digest, e.edge.Index)
if err != nil {
logrus.Error(errors.Wrap(err, "invalid query response")) // make the build fail for this error
bklog.G(context.TODO()).Error(errors.Wrap(err, "invalid query response")) // make the build fail for this error
} else {
for _, k := range keys {
records, err := e.op.Cache().Records(k)
if err != nil {
logrus.Errorf("error receiving cache records: %v", err)
bklog.G(context.TODO()).Errorf("error receiving cache records: %v", err)
continue
}
@@ -573,7 +573,7 @@ func (e *edge) recalcCurrentState() {
records, err := e.op.Cache().Records(mergedKey)
if err != nil {
logrus.Errorf("error receiving cache records: %v", err)
bklog.G(context.TODO()).Errorf("error receiving cache records: %v", err)
continue
}
@@ -665,7 +665,7 @@ func (e *edge) recalcCurrentState() {
if len(openKeys) == 0 {
e.state = edgeStatusCacheSlow
if debugScheduler {
logrus.Debugf("upgrade to cache-slow because no open keys")
bklog.G(context.TODO()).Debugf("upgrade to cache-slow because no open keys")
}
}
}
@@ -704,7 +704,7 @@ func (e *edge) respondToIncoming(incoming []pipe.Sender, allPipes []pipe.Receive
}
if debugScheduler {
logrus.Debugf("status state=%s cancomplete=%v hasouts=%v noPossibleCache=%v depsCacheFast=%v keys=%d cacheRecords=%d", e.state, allIncomingCanComplete, e.hasActiveOutgoing, e.noCacheMatchPossible, e.allDepsCompletedCacheFast, len(e.keys), len(e.cacheRecords))
bklog.G(context.TODO()).Debugf("status state=%s cancomplete=%v hasouts=%v noPossibleCache=%v depsCacheFast=%v keys=%d cacheRecords=%d", e.state, allIncomingCanComplete, e.hasActiveOutgoing, e.noCacheMatchPossible, e.allDepsCompletedCacheFast, len(e.keys), len(e.cacheRecords))
}
if allIncomingCanComplete && e.hasActiveOutgoing {
@@ -876,10 +876,10 @@ func (e *edge) loadCache(ctx context.Context) (interface{}, error) {
rec := getBestResult(recs)
e.cacheRecordsLoaded[rec.ID] = struct{}{}
logrus.Debugf("load cache for %s with %s", e.edge.Vertex.Name(), rec.ID)
bklog.G(ctx).Debugf("load cache for %s with %s", e.edge.Vertex.Name(), rec.ID)
res, err := e.op.LoadCache(ctx, rec)
if err != nil {
logrus.Debugf("load cache for %s err: %v", e.edge.Vertex.Name(), err)
bklog.G(ctx).Debugf("load cache for %s err: %v", e.edge.Vertex.Name(), err)
return nil, errors.Wrap(err, "failed to load cache")
}

6
solver/llbsolver/bridge.go
View File

@@ -18,11 +18,11 @@ import (
"github.com/moby/buildkit/solver/errdefs"
llberrdefs "github.com/moby/buildkit/solver/llbsolver/errdefs"
"github.com/moby/buildkit/solver/pb"
"github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/util/flightcontrol"
"github.com/moby/buildkit/worker"
digest "github.com/opencontainers/go-digest"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
type llbBridge struct {
@@ -69,7 +69,7 @@ func (b *llbBridge) loadResult(ctx context.Context, def *pb.Definition, cacheImp
cmNew, err = ci.Resolve(ctx, desc, cmID, w)
return err
}); err != nil {
logrus.Debugf("error while importing cache manifest from cmId=%s: %v", cmID, err)
bklog.G(ctx).Debugf("error while importing cache manifest from cmId=%s: %v", cmID, err)
return nil, err
}
return cmNew, nil
@@ -181,7 +181,7 @@ func (rp *resultProxy) Release(ctx context.Context) (err error) {
}
if rp.v != nil {
if rp.released {
logrus.Warnf("release of already released result")
bklog.G(ctx).Warnf("release of already released result")
}
rerr := rp.v.Release(ctx)
if err != nil {

8
solver/llbsolver/errdefs/exec.go
View File

@@ -5,7 +5,7 @@ import (
"runtime"
"github.com/moby/buildkit/solver"
"github.com/sirupsen/logrus"
"github.com/moby/buildkit/util/bklog"
)
// ExecError will be returned when an error is encountered when evaluating an op.
@@ -62,6 +62,10 @@ func (e *ExecError) Release() error {
}
func WithExecError(err error, inputs, mounts []solver.Result) error {
return WithExecErrorWithContext(context.TODO(), err, inputs, mounts)
}
func WithExecErrorWithContext(ctx context.Context, err error, inputs, mounts []solver.Result) error {
if err == nil {
return nil
}
@@ -73,7 +77,7 @@ func WithExecError(err error, inputs, mounts []solver.Result) error {
runtime.SetFinalizer(ee, func(e *ExecError) {
if !e.OwnerBorrowed {
e.EachRef(func(r solver.Result) error {
logrus.Warn("leaked execError detected and released")
bklog.G(ctx).Warn("leaked execError detected and released")
r.Release(context.TODO())
return nil
})

4
solver/llbsolver/file/refmanager.go
View File

@@ -7,8 +7,8 @@ import (
"github.com/moby/buildkit/session"
"github.com/moby/buildkit/snapshot"
"github.com/moby/buildkit/solver/llbsolver/ops/fileoptypes"
"github.com/moby/buildkit/util/bklog"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
func NewRefManager(cm cache.Manager) *RefManager {
@@ -41,7 +41,7 @@ func (rm *RefManager) Prepare(ctx context.Context, ref fileoptypes.Ref, readonly
if rerr != nil {
cache.CachePolicyDefault(mr)
if err := mr.Metadata().Commit(); err != nil {
logrus.Errorf("failed to reset FileOp mutable ref cachepolicy: %v", err)
bklog.G(ctx).Errorf("failed to reset FileOp mutable ref cachepolicy: %v", err)
}
mr.Release(context.TODO())
}

5
solver/llbsolver/mounts/mount.go
View File

@@ -9,6 +9,8 @@ import (
"sync"
"time"
"github.com/moby/buildkit/util/bklog"
"github.com/containerd/containerd/mount"
"github.com/containerd/containerd/pkg/userns"
"github.com/docker/docker/pkg/idtools"
@@ -24,7 +26,6 @@ import (
"github.com/moby/buildkit/util/grpcerrors"
"github.com/moby/locker"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
bolt "go.etcd.io/bbolt"
"google.golang.org/grpc/codes"
)
@@ -120,7 +121,7 @@ func (g *cacheRefGetter) getRefCacheDirNoCache(ctx context.Context, key string,
locked := false
for _, si := range sis {
if mRef, err := g.cm.GetMutable(ctx, si.ID()); err == nil {
logrus.Debugf("reusing ref for cache dir: %s", mRef.ID())
bklog.G(ctx).Debugf("reusing ref for cache dir: %s", mRef.ID())
return mRef, nil
} else if errors.Is(err, cache.ErrLocked) {
locked = true

4
solver/llbsolver/ops/exec.go
View File

@@ -22,13 +22,13 @@ import (
"github.com/moby/buildkit/solver/llbsolver/errdefs"
"github.com/moby/buildkit/solver/llbsolver/mounts"
"github.com/moby/buildkit/solver/pb"
"github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/util/progress/logs"
utilsystem "github.com/moby/buildkit/util/system"
"github.com/moby/buildkit/worker"
digest "github.com/opencontainers/go-digest"
specs "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/sync/semaphore"
)
@@ -306,7 +306,7 @@ func (e *execOp) Exec(ctx context.Context, g session.Group, inputs []solver.Resu
})
}
if err != nil {
logrus.Warn(err.Error()) // TODO: remove this with pull support
bklog.G(ctx).Warn(err.Error()) // TODO: remove this with pull support
}
meta := executor.Meta{

2
solver/llbsolver/ops/file.go
View File

@@ -452,7 +452,7 @@ func (s *FileOpSolver) getInput(ctx context.Context, idx int, inputs []fileoptyp
}
}
err = errdefs.WithExecError(err, inputRes, outputRes)
err = errdefs.WithExecErrorWithContext(ctx, err, inputRes, outputRes)
}
for _, m := range toRelease {
m.Release(context.TODO())

7
solver/progress.go
View File

@@ -5,10 +5,11 @@ import (
"io"
"time"
"github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/client"
"github.com/moby/buildkit/util/progress"
digest "github.com/opencontainers/go-digest"
"github.com/sirupsen/logrus"
)
func (j *Job) Status(ctx context.Context, ch chan *client.SolveStatus) error {
@@ -38,7 +39,7 @@ func (j *Job) Status(ctx context.Context, ch chan *client.SolveStatus) error {
case progress.Status:
vtx, ok := p.Meta("vertex")
if !ok {
logrus.Warnf("progress %s status without vertex info", p.ID)
bklog.G(ctx).Warnf("progress %s status without vertex info", p.ID)
continue
}
vs := &client.VertexStatus{
@@ -55,7 +56,7 @@ func (j *Job) Status(ctx context.Context, ch chan *client.SolveStatus) error {
case client.VertexLog:
vtx, ok := p.Meta("vertex")
if !ok {
logrus.Warnf("progress %s log without vertex info", p.ID)
bklog.G(ctx).Warnf("progress %s log without vertex info", p.ID)
continue
}
v.Vertex = vtx.(digest.Digest)

5
solver/result.go
View File

@@ -5,8 +5,9 @@ import (
"sync"
"sync/atomic"
"github.com/moby/buildkit/util/bklog"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
// SharedResult is a result that can be cloned
@@ -48,7 +49,7 @@ type splitResult struct {
func (r *splitResult) Release(ctx context.Context) error {
if atomic.AddInt64(&r.released, 1) > 1 {
err := errors.Errorf("releasing already released reference %+v", r.Result.ID())
logrus.Error(err)
bklog.G(ctx).Error(err)
return err
}
if atomic.AddInt64(r.sem, 1) == 2 {

31
solver/scheduler.go
View File

@@ -6,9 +6,9 @@ import (
"sync"
"github.com/moby/buildkit/solver/internal/pipe"
"github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/util/cond"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
var debugScheduler = false // TODO: replace with logs in build trace
@@ -171,9 +171,9 @@ postUnpark:
origEdge := e.index.LoadOrStore(k, e)
if origEdge != nil {
if e.isDep(origEdge) || origEdge.isDep(e) {
logrus.Debugf("skip merge due to dependency")
bklog.G(context.TODO()).Debugf("skip merge due to dependency")
} else {
logrus.Debugf("merging edge %s to %s\n", e.edge.Vertex.Name(), origEdge.edge.Vertex.Name())
bklog.G(context.TODO()).Debugf("merging edge %s to %s\n", e.edge.Vertex.Name(), origEdge.edge.Vertex.Name())
if s.mergeTo(origEdge, e) {
s.ef.setEdge(e.edge, origEdge)
}
@@ -355,7 +355,7 @@ func (pf *pipeFactory) NewInputRequest(ee Edge, req *edgeRequest) pipe.Receiver
}
p := pf.s.newPipe(target, pf.e, pipe.Request{Payload: req})
if debugScheduler {
logrus.Debugf("> newPipe %s %p desiredState=%s", ee.Vertex.Name(), p, req.desiredState)
bklog.G(context.TODO()).Debugf("> newPipe %s %p desiredState=%s", ee.Vertex.Name(), p, req.desiredState)
}
return p.Receiver
}
@@ -363,32 +363,34 @@ func (pf *pipeFactory) NewInputRequest(ee Edge, req *edgeRequest) pipe.Receiver
func (pf *pipeFactory) NewFuncRequest(f func(context.Context) (interface{}, error)) pipe.Receiver {
p := pf.s.newRequestWithFunc(pf.e, f)
if debugScheduler {
logrus.Debugf("> newFunc %p", p)
bklog.G(context.TODO()).Debugf("> newFunc %p", p)
}
return p
}
func debugSchedulerPreUnpark(e *edge, inc []pipe.Sender, updates, allPipes []pipe.Receiver) {
logrus.Debugf(">> unpark %s req=%d upt=%d out=%d state=%s %s", e.edge.Vertex.Name(), len(inc), len(updates), len(allPipes), e.state, e.edge.Vertex.Digest())
log := bklog.G(context.TODO())
log.Debugf(">> unpark %s req=%d upt=%d out=%d state=%s %s", e.edge.Vertex.Name(), len(inc), len(updates), len(allPipes), e.state, e.edge.Vertex.Digest())
for i, dep := range e.deps {
des := edgeStatusInitial
if dep.req != nil {
des = dep.req.Request().(*edgeRequest).desiredState
}
logrus.Debugf(":: dep%d %s state=%s des=%s keys=%d hasslowcache=%v preprocessfunc=%v", i, e.edge.Vertex.Inputs()[i].Vertex.Name(), dep.state, des, len(dep.keys), e.slowCacheFunc(dep) != nil, e.preprocessFunc(dep) != nil)
log.Debugf(":: dep%d %s state=%s des=%s keys=%d hasslowcache=%v preprocessfunc=%v", i, e.edge.Vertex.Inputs()[i].Vertex.Name(), dep.state, des, len(dep.keys), e.slowCacheFunc(dep) != nil, e.preprocessFunc(dep) != nil)
}
for i, in := range inc {
req := in.Request()
logrus.Debugf("> incoming-%d: %p dstate=%s canceled=%v", i, in, req.Payload.(*edgeRequest).desiredState, req.Canceled)
log.Debugf("> incoming-%d: %p dstate=%s canceled=%v", i, in, req.Payload.(*edgeRequest).desiredState, req.Canceled)
}
for i, up := range updates {
if up == e.cacheMapReq {
logrus.Debugf("> update-%d: %p cacheMapReq complete=%v", i, up, up.Status().Completed)
log.Debugf("> update-%d: %p cacheMapReq complete=%v", i, up, up.Status().Completed)
} else if up == e.execReq {
logrus.Debugf("> update-%d: %p execReq complete=%v", i, up, up.Status().Completed)
log.Debugf("> update-%d: %p execReq complete=%v", i, up, up.Status().Completed)
} else {
st, ok := up.Status().Value.(*edgeState)
if ok {
@@ -396,17 +398,18 @@ func debugSchedulerPreUnpark(e *edge, inc []pipe.Sender, updates, allPipes []pip
if dep, ok := e.depRequests[up]; ok {
index = int(dep.index)
}
logrus.Debugf("> update-%d: %p input-%d keys=%d state=%s", i, up, index, len(st.keys), st.state)
log.Debugf("> update-%d: %p input-%d keys=%d state=%s", i, up, index, len(st.keys), st.state)
} else {
logrus.Debugf("> update-%d: unknown", i)
log.Debugf("> update-%d: unknown", i)
}
}
}
}
func debugSchedulerPostUnpark(e *edge, inc []pipe.Sender) {
log := bklog.G(context.TODO())
for i, in := range inc {
logrus.Debugf("< incoming-%d: %p completed=%v", i, in, in.Status().Completed)
log.Debugf("< incoming-%d: %p completed=%v", i, in, in.Status().Completed)
}
logrus.Debugf("<< unpark %s\n", e.edge.Vertex.Name())
log.Debugf("<< unpark %s\n", e.edge.Vertex.Name())
}

5
source/git/gitsource.go
View File

@@ -17,6 +17,8 @@ import (
"strconv"
"strings"
"github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/cache"
"github.com/moby/buildkit/cache/metadata"
"github.com/moby/buildkit/client"
@@ -30,7 +32,6 @@ import (
"github.com/moby/buildkit/util/progress/logs"
"github.com/moby/locker"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
bolt "go.etcd.io/bbolt"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
@@ -86,7 +87,7 @@ func (gs *gitSource) mountRemote(ctx context.Context, remote string, auth []stri
if err != nil {
if errors.Is(err, cache.ErrLocked) {
// should never really happen as no other function should access this metadata, but lets be graceful
logrus.Warnf("mutable ref for %s %s was locked: %v", redactCredentials(remote), si.ID(), err)
bklog.G(ctx).Warnf("mutable ref for %s %s was locked: %v", redactCredentials(remote), si.ID(), err)
continue
}
return "", nil, errors.Wrapf(err, "failed to get mutable ref for %s", redactCredentials(remote))

11
source/local/local.go
View File

@@ -6,6 +6,8 @@ import (
"fmt"
"time"
"github.com/moby/buildkit/util/bklog"
"github.com/docker/docker/pkg/idtools"
"github.com/moby/buildkit/cache"
"github.com/moby/buildkit/cache/contenthash"
@@ -19,7 +21,6 @@ import (
"github.com/moby/buildkit/util/progress"
digest "github.com/opencontainers/go-digest"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/tonistiigi/fsutil"
fstypes "github.com/tonistiigi/fsutil/types"
bolt "go.etcd.io/bbolt"
@@ -119,7 +120,7 @@ func (ls *localSourceHandler) snapshot(ctx context.Context, s session.Group, cal
}
for _, si := range sis {
if m, err := ls.cm.GetMutable(ctx, si.ID()); err == nil {
logrus.Debugf("reusing ref for local: %s", m.ID())
bklog.G(ctx).Debugf("reusing ref for local: %s", m.ID())
mutable = m
break
}
@@ -131,7 +132,7 @@ func (ls *localSourceHandler) snapshot(ctx context.Context, s session.Group, cal
return nil, err
}
mutable = m
logrus.Debugf("new ref for local: %s", mutable.ID())
bklog.G(ctx).Debugf("new ref for local: %s", mutable.ID())
}
defer func() {
@@ -139,7 +140,7 @@ func (ls *localSourceHandler) snapshot(ctx context.Context, s session.Group, cal
// on error remove the record as checksum update is in undefined state
cache.CachePolicyDefault(mutable)
if err := mutable.Metadata().Commit(); err != nil {
logrus.Errorf("failed to reset mutable cachepolicy: %v", err)
bklog.G(ctx).Errorf("failed to reset mutable cachepolicy: %v", err)
}
contenthash.ClearCacheContext(mutable.Metadata())
go mutable.Release(context.TODO())
@@ -233,7 +234,7 @@ func (ls *localSourceHandler) snapshot(ctx context.Context, s session.Group, cal
}); err != nil {
return nil, err
}
logrus.Debugf("saved %s as %s", mutable.ID(), sharedKey)
bklog.G(ctx).Debugf("saved %s as %s", mutable.ID(), sharedKey)
}
snap, err := mutable.Commit(ctx)

46
util/bklog/log.go Normal file
View File

@@ -0,0 +1,46 @@
package bklog
import (
"context"
"github.com/sirupsen/logrus"
"go.opentelemetry.io/otel/trace"
)
var (
G = GetLogger
L = logrus.NewEntry(logrus.StandardLogger())
)
type (
loggerKey struct{}
)
// WithLogger returns a new context with the provided logger. Use in
// combination with logger.WithField(s) for great effect.
func WithLogger(ctx context.Context, logger *logrus.Entry) context.Context {
return context.WithValue(ctx, loggerKey{}, logger)
}
// GetLogger retrieves the current logger from the context. If no logger is
// available, the default logger is returned.
func GetLogger(ctx context.Context) (l *logrus.Entry) {
logger := ctx.Value(loggerKey{})
if logger != nil {
l = logger.(*logrus.Entry)
} else {
l = L
}
spanContext := trace.SpanFromContext(ctx).SpanContext()
if spanContext.IsValid() {
return l.WithFields(logrus.Fields{
"traceID": spanContext.TraceID(),
"spanID": spanContext.SpanID(),
})
}
return l
}

28
util/log/log.go
View File

@@ -1,28 +0,0 @@
package log
import (
"context"
"github.com/containerd/containerd/log"
"github.com/sirupsen/logrus"
"go.opentelemetry.io/otel/trace"
)
func init() {
log.G = GetLogger
}
func GetLogger(ctx context.Context) *logrus.Entry {
l := log.GetLogger(ctx)
spanContext := trace.SpanFromContext(ctx).SpanContext()
if spanContext.IsValid() {
return l.WithFields(logrus.Fields{
"traceID": spanContext.TraceID(),
"spanID": spanContext.SpanID(),
})
}
return l
}

2
util/resolver/authorizer.go
View File

@@ -11,12 +11,12 @@ import (
"time"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/remotes/docker"
"github.com/containerd/containerd/remotes/docker/auth"
remoteserrors "github.com/containerd/containerd/remotes/errors"
"github.com/moby/buildkit/session"
sessionauth "github.com/moby/buildkit/session/auth"
log "github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/util/flightcontrol"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"

2
util/tracing/tracing.go
View File

@@ -6,6 +6,7 @@ import (
"io"
"net/http"
"github.com/moby/buildkit/util/bklog"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/propagation"
@@ -22,6 +23,7 @@ func StartSpan(ctx context.Context, operationName string, opts ...trace.SpanStar
tracer = parent.TracerProvider().Tracer("")
}
ctx, span := tracer.Start(ctx, operationName, opts...)
ctx = bklog.WithLogger(ctx, bklog.GetLogger(ctx).WithField("span", operationName))
return span, ctx
}

3
util/winlayers/differ.go
View File

@@ -14,12 +14,13 @@ import (
"github.com/containerd/containerd/content"
"github.com/containerd/containerd/diff"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/mount"
digest "github.com/opencontainers/go-digest"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
log "github.com/moby/buildkit/util/bklog"
)
const (

10
worker/base/worker.go
View File

@@ -42,13 +42,13 @@ import (
"github.com/moby/buildkit/source/http"
"github.com/moby/buildkit/source/local"
"github.com/moby/buildkit/util/archutil"
"github.com/moby/buildkit/util/bklog"
"github.com/moby/buildkit/util/progress"
"github.com/moby/buildkit/util/progress/controller"
"github.com/moby/buildkit/worker"
digest "github.com/opencontainers/go-digest"
specs "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
bolt "go.etcd.io/bbolt"
"golang.org/x/sync/semaphore"
)
@@ -89,7 +89,7 @@ type Worker struct {
}
// NewWorker instantiates a local worker
func NewWorker(opt WorkerOpt) (*Worker, error) {
func NewWorker(ctx context.Context, opt WorkerOpt) (*Worker, error) {
imageRefChecker := imagerefchecker.New(imagerefchecker.Opt{
ImageStore: opt.ImageStore,
ContentStore: opt.ContentStore,
@@ -139,7 +139,7 @@ func NewWorker(opt WorkerOpt) (*Worker, error) {
}
sm.Register(gs)
} else {
logrus.Warnf("git source cannot be enabled: %v", err)
bklog.G(ctx).Warnf("git source cannot be enabled: %v", err)
}
hs, err := http.NewSource(http.Opt{
@@ -171,12 +171,12 @@ func NewWorker(opt WorkerOpt) (*Worker, error) {
return nil, err
}
leases, err := opt.LeaseManager.List(context.TODO(), "labels.\"buildkit/lease.temporary\"")
leases, err := opt.LeaseManager.List(ctx, "labels.\"buildkit/lease.temporary\"")
if err != nil {
return nil, err
}
for _, l := range leases {
opt.LeaseManager.Delete(context.TODO(), l)
opt.LeaseManager.Delete(ctx, l)
}
return &Worker{

5
worker/containerd/containerd_test.go
View File

@@ -3,6 +3,7 @@
package containerd
import (
"context"
"io/ioutil"
"os"
"testing"
@@ -44,7 +45,7 @@ func checkRequirement(t *testing.T) {
func testContainerdWorkerExec(t *testing.T, sb integration.Sandbox) {
workerOpt, cleanupWorkerOpt := newWorkerOpt(t, sb.ContainerdAddress())
defer cleanupWorkerOpt()
w, err := base.NewWorker(workerOpt)
w, err := base.NewWorker(context.TODO(), workerOpt)
require.NoError(t, err)
tests.TestWorkerExec(t, w)
@@ -53,7 +54,7 @@ func testContainerdWorkerExec(t *testing.T, sb integration.Sandbox) {
func testContainerdWorkerExecFailures(t *testing.T, sb integration.Sandbox) {
workerOpt, cleanupWorkerOpt := newWorkerOpt(t, sb.ContainerdAddress())
defer cleanupWorkerOpt()
w, err := base.NewWorker(workerOpt)
w, err := base.NewWorker(context.TODO(), workerOpt)
require.NoError(t, err)
tests.TestWorkerExecFailures(t, w)

8
worker/runc/runc_test.go
View File

@@ -64,7 +64,7 @@ func TestRuncWorker(t *testing.T) {
workerOpt, cleanupWorkerOpt := newWorkerOpt(t, oci.ProcessSandbox)
defer cleanupWorkerOpt()
w, err := base.NewWorker(workerOpt)
w, err := base.NewWorker(context.TODO(), workerOpt)
require.NoError(t, err)
ctx := tests.NewCtx("buildkit-test")
@@ -186,7 +186,7 @@ func TestRuncWorkerNoProcessSandbox(t *testing.T) {
workerOpt, cleanupWorkerOpt := newWorkerOpt(t, oci.NoProcessSandbox)
defer cleanupWorkerOpt()
w, err := base.NewWorker(workerOpt)
w, err := base.NewWorker(context.TODO(), workerOpt)
require.NoError(t, err)
ctx := tests.NewCtx("buildkit-test")
@@ -217,7 +217,7 @@ func TestRuncWorkerExec(t *testing.T) {
workerOpt, cleanupWorkerOpt := newWorkerOpt(t, oci.ProcessSandbox)
defer cleanupWorkerOpt()
w, err := base.NewWorker(workerOpt)
w, err := base.NewWorker(context.TODO(), workerOpt)
require.NoError(t, err)
tests.TestWorkerExec(t, w)
@@ -229,7 +229,7 @@ func TestRuncWorkerExecFailures(t *testing.T) {
workerOpt, cleanupWorkerOpt := newWorkerOpt(t, oci.ProcessSandbox)
defer cleanupWorkerOpt()
w, err := base.NewWorker(workerOpt)
w, err := base.NewWorker(context.TODO(), workerOpt)
require.NoError(t, err)
tests.TestWorkerExecFailures(t, w)