mirror of https://github.com/docker/compose.git
remove support of Synchronize File Shares integration with Docker Desktop
Signed-off-by: Guillaume Lours <705411+glours@users.noreply.github.com>
This commit is contained in:
Guillaume Lours
Nicolas De loof
parent
f46689a75e
commit
f5491328bb
2
go.mod
2
go.mod
|
|
@ -37,7 +37,6 @@ require (
|
|||
github.com/opencontainers/go-digest v1.0.0
|
||||
github.com/opencontainers/image-spec v1.1.1
|
||||
github.com/otiai10/copy v1.14.1
|
||||
github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc
|
||||
github.com/sirupsen/logrus v1.9.3
|
||||
github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966
|
||||
github.com/spf13/cobra v1.9.1
|
||||
|
|
@ -193,7 +192,6 @@ require (
|
|||
google.golang.org/genproto/googleapis/api v0.0.0-20250218202821-56aae31c358a // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20250218202821-56aae31c358a // indirect
|
||||
google.golang.org/protobuf v1.36.5 // indirect
|
||||
gopkg.in/cenkalti/backoff.v1 v1.1.0 // indirect
|
||||
gopkg.in/inf.v0 v0.9.1 // indirect
|
||||
gopkg.in/ini.v1 v1.67.0 // indirect
|
||||
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||
|
|
|
|||
5
go.sum
5
go.sum
|
|
@ -420,8 +420,6 @@ github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsT
|
|||
github.com/prometheus/procfs v0.0.3/go.mod h1:4A/X28fw3Fc593LaREMrKMqOKvUAntwMDaekg4FpcdQ=
|
||||
github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
|
||||
github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
|
||||
github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc h1:zAsgcP8MhzAbhMnB1QQ2O7ZhWYVGYSR2iVcjzQuPV+o=
|
||||
github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc/go.mod h1:S8xSOnV3CgpNrWd0GQ/OoQfMtlg2uPRSuTzcSGrzwK8=
|
||||
github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
|
||||
github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
|
||||
github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
|
||||
|
|
@ -561,7 +559,6 @@ golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73r
|
|||
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20191116160921-f9c825593386/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
|
||||
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
|
||||
|
|
@ -637,8 +634,6 @@ google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwl
|
|||
google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
|
||||
gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U=
|
||||
gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
|
||||
gopkg.in/cenkalti/backoff.v1 v1.1.0 h1:Arh75ttbsvlpVA7WtVpH4u9h6Zl46xuptxqLxPiSo4Y=
|
||||
gopkg.in/cenkalti/backoff.v1 v1.1.0/go.mod h1:J6Vskwqd+OMVJl8C33mmtxTBs2gyzfv7UDAkHu8BrjI=
|
||||
gopkg.in/cenkalti/backoff.v2 v2.2.1 h1:eJ9UAg01/HIHG987TwxvnzK2MgxXq97YY6rYDpY9aII=
|
||||
gopkg.in/cenkalti/backoff.v2 v2.2.1/go.mod h1:S0QdOvT2AlerfSBkp0O+dk+bbIMaNbEmVk876gPCthU=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
|
|
|
|||
|
|
@ -17,10 +17,8 @@
|
|||
package desktop
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
|
|
@ -29,7 +27,6 @@ import (
|
|||
|
||||
"github.com/docker/compose/v2/internal"
|
||||
"github.com/docker/compose/v2/internal/memnet"
|
||||
"github.com/r3labs/sse"
|
||||
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
|
||||
)
|
||||
|
||||
|
|
@ -130,212 +127,6 @@ func (c *Client) FeatureFlags(ctx context.Context) (FeatureFlagResponse, error)
|
|||
return ret, nil
|
||||
}
|
||||
|
||||
type GetFileSharesConfigResponse struct {
|
||||
Active bool `json:"active"`
|
||||
Compose struct {
|
||||
ManageBindMounts bool `json:"manageBindMounts"`
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) GetFileSharesConfig(ctx context.Context) (*GetFileSharesConfigResponse, error) {
|
||||
req, err := c.newRequest(ctx, http.MethodGet, "/mutagen/file-shares/config", http.NoBody)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resp, err := c.client.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
_ = resp.Body.Close()
|
||||
}()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, newHTTPStatusCodeError(resp)
|
||||
}
|
||||
|
||||
var ret GetFileSharesConfigResponse
|
||||
if err := json.NewDecoder(resp.Body).Decode(&ret); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &ret, nil
|
||||
}
|
||||
|
||||
type CreateFileShareRequest struct {
|
||||
HostPath string `json:"hostPath"`
|
||||
Labels map[string]string `json:"labels,omitempty"`
|
||||
}
|
||||
|
||||
type CreateFileShareResponse struct {
|
||||
FileShareID string `json:"fileShareID"`
|
||||
}
|
||||
|
||||
func (c *Client) CreateFileShare(ctx context.Context, r CreateFileShareRequest) (*CreateFileShareResponse, error) {
|
||||
rawBody, err := json.Marshal(r)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to marshal request: %w", err)
|
||||
}
|
||||
|
||||
req, err := c.newRequest(ctx, http.MethodPost, "/mutagen/file-shares", bytes.NewReader(rawBody))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
resp, err := c.client.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
_ = resp.Body.Close()
|
||||
}()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
errBody, _ := io.ReadAll(resp.Body)
|
||||
return nil, fmt.Errorf("unexpected status code %d: %s", resp.StatusCode, string(errBody))
|
||||
}
|
||||
|
||||
var ret CreateFileShareResponse
|
||||
if err := json.NewDecoder(resp.Body).Decode(&ret); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &ret, nil
|
||||
}
|
||||
|
||||
type FileShareReceiverState struct {
|
||||
TotalReceivedSize uint64 `json:"totalReceivedSize"`
|
||||
}
|
||||
|
||||
type FileShareEndpoint struct {
|
||||
Path string `json:"path"`
|
||||
TotalFileSize uint64 `json:"totalFileSize,omitempty"`
|
||||
StagingProgress *FileShareReceiverState `json:"stagingProgress"`
|
||||
}
|
||||
|
||||
type FileShareSession struct {
|
||||
SessionID string `json:"identifier"`
|
||||
Alpha FileShareEndpoint `json:"alpha"`
|
||||
Beta FileShareEndpoint `json:"beta"`
|
||||
Labels map[string]string `json:"labels"`
|
||||
Status string `json:"status"`
|
||||
}
|
||||
|
||||
func (c *Client) ListFileShares(ctx context.Context) ([]FileShareSession, error) {
|
||||
req, err := c.newRequest(ctx, http.MethodGet, "/mutagen/file-shares", http.NoBody)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resp, err := c.client.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
_ = resp.Body.Close()
|
||||
}()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, newHTTPStatusCodeError(resp)
|
||||
}
|
||||
|
||||
var ret []FileShareSession
|
||||
if err := json.NewDecoder(resp.Body).Decode(&ret); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return ret, nil
|
||||
}
|
||||
|
||||
func (c *Client) DeleteFileShare(ctx context.Context, id string) error {
|
||||
req, err := c.newRequest(ctx, http.MethodDelete, "/mutagen/file-shares/"+id, http.NoBody)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
resp, err := c.client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
_ = resp.Body.Close()
|
||||
}()
|
||||
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return newHTTPStatusCodeError(resp)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type EventMessage[T any] struct {
|
||||
Value T
|
||||
Error error
|
||||
}
|
||||
|
||||
func newHTTPStatusCodeError(resp *http.Response) error {
|
||||
r := io.LimitReader(resp.Body, 2048)
|
||||
body, err := io.ReadAll(r)
|
||||
if err != nil {
|
||||
return fmt.Errorf("http status code %d", resp.StatusCode)
|
||||
}
|
||||
return fmt.Errorf("http status code %d: %s", resp.StatusCode, string(body))
|
||||
}
|
||||
|
||||
func (c *Client) StreamFileShares(ctx context.Context) (<-chan EventMessage[[]FileShareSession], error) {
|
||||
req, err := c.newRequest(ctx, http.MethodGet, "/mutagen/file-shares/stream", http.NoBody)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resp, err := c.client.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
_ = resp.Body.Close()
|
||||
return nil, newHTTPStatusCodeError(resp)
|
||||
}
|
||||
|
||||
events := make(chan EventMessage[[]FileShareSession])
|
||||
go func(ctx context.Context) {
|
||||
defer func() {
|
||||
_ = resp.Body.Close()
|
||||
close(events)
|
||||
}()
|
||||
if err := readEvents(ctx, resp.Body, events); err != nil {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case events <- EventMessage[[]FileShareSession]{Error: err}:
|
||||
}
|
||||
}
|
||||
}(ctx)
|
||||
return events, nil
|
||||
}
|
||||
|
||||
func readEvents[T any](ctx context.Context, r io.Reader, events chan<- EventMessage[T]) error {
|
||||
eventReader := sse.NewEventStreamReader(r)
|
||||
for {
|
||||
msg, err := eventReader.ReadEvent()
|
||||
if errors.Is(err, io.EOF) {
|
||||
return nil
|
||||
} else if err != nil {
|
||||
return fmt.Errorf("reading events: %w", err)
|
||||
}
|
||||
msg = bytes.TrimPrefix(msg, []byte("data: "))
|
||||
|
||||
var event T
|
||||
if err := json.Unmarshal(msg, &event); err != nil {
|
||||
return err
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return context.Cause(ctx)
|
||||
case events <- EventMessage[T]{Value: event}:
|
||||
// event was sent to channel, read next
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) newRequest(ctx context.Context, method, path string, body io.Reader) (*http.Request, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, method, backendURL(path), body)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -1,384 +0,0 @@
|
|||
/*
|
||||
Copyright 2024 Docker Compose CLI authors
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package desktop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/docker/compose/v2/internal/paths"
|
||||
"github.com/docker/compose/v2/pkg/api"
|
||||
"github.com/docker/compose/v2/pkg/progress"
|
||||
"github.com/docker/go-units"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// fileShareProgressID is the identifier used for the root grouping of file
|
||||
// share events in the progress writer.
|
||||
const fileShareProgressID = "Synchronized File Shares"
|
||||
|
||||
// RemoveFileSharesForProject removes any Synchronized File Shares that were
|
||||
// created by Compose for this project in the past if possible.
|
||||
//
|
||||
// Errors are not propagated; they are only sent to the progress writer.
|
||||
func RemoveFileSharesForProject(ctx context.Context, c *Client, projectName string) {
|
||||
w := progress.ContextWriter(ctx)
|
||||
|
||||
existing, err := c.ListFileShares(ctx)
|
||||
if err != nil {
|
||||
w.TailMsgf("Synchronized File Shares not removed due to error: %v", err)
|
||||
return
|
||||
}
|
||||
// filter the list first, so we can early return and not show the event if
|
||||
// there's no sessions to clean up
|
||||
var toRemove []FileShareSession
|
||||
for _, share := range existing {
|
||||
if share.Labels["com.docker.compose.project"] == projectName {
|
||||
toRemove = append(toRemove, share)
|
||||
}
|
||||
}
|
||||
if len(toRemove) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
w.Event(progress.NewEvent(fileShareProgressID, progress.Working, "Removing"))
|
||||
rootResult := progress.Done
|
||||
defer func() {
|
||||
w.Event(progress.NewEvent(fileShareProgressID, rootResult, ""))
|
||||
}()
|
||||
for _, share := range toRemove {
|
||||
shareID := share.Labels["com.docker.desktop.mutagen.file-share.id"]
|
||||
if shareID == "" {
|
||||
w.Event(progress.Event{
|
||||
ID: share.Alpha.Path,
|
||||
ParentID: fileShareProgressID,
|
||||
Status: progress.Warning,
|
||||
StatusText: "Invalid",
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
w.Event(progress.Event{
|
||||
ID: share.Alpha.Path,
|
||||
ParentID: fileShareProgressID,
|
||||
Status: progress.Working,
|
||||
})
|
||||
|
||||
var status progress.EventStatus
|
||||
var statusText string
|
||||
if err := c.DeleteFileShare(ctx, shareID); err != nil {
|
||||
// TODO(milas): Docker Desktop is doing weird things with error responses,
|
||||
// once fixed, we can return proper error types from the client
|
||||
if strings.Contains(err.Error(), "file share in use") {
|
||||
status = progress.Warning
|
||||
statusText = "Resource is still in use"
|
||||
if rootResult != progress.Error {
|
||||
// error takes precedence over warning
|
||||
rootResult = progress.Warning
|
||||
}
|
||||
} else {
|
||||
logrus.Debugf("Error deleting file share %q: %v", shareID, err)
|
||||
status = progress.Error
|
||||
rootResult = progress.Error
|
||||
}
|
||||
} else {
|
||||
logrus.Debugf("Deleted file share: %s", shareID)
|
||||
status = progress.Done
|
||||
}
|
||||
|
||||
w.Event(progress.Event{
|
||||
ID: share.Alpha.Path,
|
||||
ParentID: fileShareProgressID,
|
||||
Status: status,
|
||||
StatusText: statusText,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// FileShareManager maps between Compose bind mounts and Desktop File Shares
|
||||
// state.
|
||||
type FileShareManager struct {
|
||||
mu sync.Mutex
|
||||
cli *Client
|
||||
projectName string
|
||||
hostPaths []string
|
||||
// state holds session status keyed by file share ID.
|
||||
state map[string]*FileShareSession
|
||||
}
|
||||
|
||||
func NewFileShareManager(cli *Client, projectName string, hostPaths []string) *FileShareManager {
|
||||
return &FileShareManager{
|
||||
cli: cli,
|
||||
projectName: projectName,
|
||||
hostPaths: hostPaths,
|
||||
state: make(map[string]*FileShareSession),
|
||||
}
|
||||
}
|
||||
|
||||
// EnsureExists looks for existing File Shares or creates new ones for the
|
||||
// host paths.
|
||||
//
|
||||
// This function blocks until each share reaches steady state, at which point
|
||||
// flow can continue.
|
||||
func (m *FileShareManager) EnsureExists(ctx context.Context) (err error) {
|
||||
w := progress.ContextWriter(ctx)
|
||||
w.Event(progress.NewEvent(fileShareProgressID, progress.Working, ""))
|
||||
defer func() {
|
||||
if err != nil {
|
||||
w.Event(progress.NewEvent(fileShareProgressID, progress.Error, ""))
|
||||
} else {
|
||||
w.Event(progress.NewEvent(fileShareProgressID, progress.Done, ""))
|
||||
}
|
||||
}()
|
||||
|
||||
wait := &waiter{
|
||||
shareIDs: make(map[string]struct{}),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
|
||||
handler := m.eventHandler(w, wait)
|
||||
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
// stream session events to update internal state for project
|
||||
monitorErr := make(chan error, 1)
|
||||
go func() {
|
||||
defer close(monitorErr)
|
||||
if err := m.watch(ctx, handler); err != nil && ctx.Err() == nil {
|
||||
monitorErr <- err
|
||||
}
|
||||
}()
|
||||
|
||||
if err := m.initialize(ctx, wait, handler); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
waitCh := wait.start()
|
||||
if waitCh != nil {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return context.Cause(ctx)
|
||||
case err := <-monitorErr:
|
||||
if err != nil {
|
||||
return fmt.Errorf("watching file share sessions: %w", err)
|
||||
} else if ctx.Err() == nil {
|
||||
// this indicates a bug - it should not stop w/o an error if the context is still active
|
||||
return errors.New("file share session watch stopped unexpectedly")
|
||||
}
|
||||
case <-wait.start():
|
||||
// everything is done
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// initialize finds existing shares or creates new ones for the host paths.
|
||||
//
|
||||
// Once a share is found/created, its progress is monitored via the watch.
|
||||
func (m *FileShareManager) initialize(ctx context.Context, wait *waiter, handler func(FileShareSession)) error {
|
||||
// the watch is already running in the background, so the lock is taken
|
||||
// throughout to prevent interleaving writes
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
existing, err := m.cli.ListFileShares(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, path := range m.hostPaths {
|
||||
var fileShareID string
|
||||
var fss *FileShareSession
|
||||
|
||||
if fss = findExistingShare(path, existing); fss != nil {
|
||||
fileShareID = fss.Beta.Path
|
||||
logrus.Debugf("Found existing suitable file share %s for path %q [%s]", fileShareID, path, fss.Alpha.Path)
|
||||
wait.addShare(fileShareID)
|
||||
handler(*fss)
|
||||
continue
|
||||
} else {
|
||||
req := CreateFileShareRequest{
|
||||
HostPath: path,
|
||||
Labels: map[string]string{
|
||||
"com.docker.compose.project": m.projectName,
|
||||
},
|
||||
}
|
||||
createResp, err := m.cli.CreateFileShare(ctx, req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("creating file share: %w", err)
|
||||
}
|
||||
fileShareID = createResp.FileShareID
|
||||
fss = m.state[fileShareID]
|
||||
logrus.Debugf("Created file share %s for path %q", fileShareID, path)
|
||||
}
|
||||
wait.addShare(fileShareID)
|
||||
if fss != nil {
|
||||
handler(*fss)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *FileShareManager) watch(ctx context.Context, handler func(FileShareSession)) error {
|
||||
events, err := m.cli.StreamFileShares(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("streaming file shares: %w", err)
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case event := <-events:
|
||||
if event.Error != nil {
|
||||
return fmt.Errorf("reading file share events: %w", event.Error)
|
||||
}
|
||||
// closure for lock
|
||||
func() {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for _, fss := range event.Value {
|
||||
handler(fss)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// eventHandler updates internal state, keeps track of in-progress syncs, and
|
||||
// prints relevant events to progress.
|
||||
func (m *FileShareManager) eventHandler(w progress.Writer, wait *waiter) func(fss FileShareSession) {
|
||||
return func(fss FileShareSession) {
|
||||
fileShareID := fss.Beta.Path
|
||||
|
||||
shouldPrint := wait.isWatching(fileShareID)
|
||||
forProject := fss.Labels[api.ProjectLabel] == m.projectName
|
||||
|
||||
if shouldPrint || forProject {
|
||||
m.state[fileShareID] = &fss
|
||||
}
|
||||
|
||||
var percent int
|
||||
var current, total int64
|
||||
if fss.Beta.StagingProgress != nil {
|
||||
current = int64(fss.Beta.StagingProgress.TotalReceivedSize)
|
||||
} else {
|
||||
current = int64(fss.Beta.TotalFileSize)
|
||||
}
|
||||
total = int64(fss.Alpha.TotalFileSize)
|
||||
if total != 0 {
|
||||
percent = int(current * 100 / total)
|
||||
}
|
||||
|
||||
var status progress.EventStatus
|
||||
var text string
|
||||
|
||||
switch {
|
||||
case strings.HasPrefix(fss.Status, "halted"):
|
||||
wait.shareDone(fileShareID)
|
||||
status = progress.Error
|
||||
case fss.Status == "watching":
|
||||
wait.shareDone(fileShareID)
|
||||
status = progress.Done
|
||||
percent = 100
|
||||
case fss.Status == "staging-beta":
|
||||
status = progress.Working
|
||||
// TODO(milas): the printer doesn't style statuses for children nicely
|
||||
text = fmt.Sprintf(" Syncing (%7s / %-7s)",
|
||||
units.HumanSize(float64(current)),
|
||||
units.HumanSize(float64(total)),
|
||||
)
|
||||
default:
|
||||
// catch-all for various other transitional statuses
|
||||
status = progress.Working
|
||||
}
|
||||
|
||||
evt := progress.Event{
|
||||
ID: fss.Alpha.Path,
|
||||
Status: status,
|
||||
Text: text,
|
||||
ParentID: fileShareProgressID,
|
||||
Current: current,
|
||||
Total: total,
|
||||
Percent: percent,
|
||||
}
|
||||
|
||||
if shouldPrint {
|
||||
w.Event(evt)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func findExistingShare(path string, existing []FileShareSession) *FileShareSession {
|
||||
for _, share := range existing {
|
||||
if paths.IsChild(share.Alpha.Path, path) {
|
||||
return &share
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type waiter struct {
|
||||
mu sync.Mutex
|
||||
shareIDs map[string]struct{}
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
func (w *waiter) addShare(fileShareID string) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.shareIDs[fileShareID] = struct{}{}
|
||||
}
|
||||
|
||||
func (w *waiter) isWatching(fileShareID string) bool {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
_, ok := w.shareIDs[fileShareID]
|
||||
return ok
|
||||
}
|
||||
|
||||
// start returns a channel to wait for any outstanding shares to be ready.
|
||||
//
|
||||
// If no shares are registered when this is called, nil is returned.
|
||||
func (w *waiter) start() <-chan struct{} {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if len(w.shareIDs) == 0 {
|
||||
return nil
|
||||
}
|
||||
if w.done == nil {
|
||||
w.done = make(chan struct{})
|
||||
}
|
||||
return w.done
|
||||
}
|
||||
|
||||
func (w *waiter) shareDone(fileShareID string) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
|
||||
delete(w.shareIDs, fileShareID)
|
||||
if len(w.shareIDs) == 0 && w.done != nil {
|
||||
close(w.done)
|
||||
w.done = nil
|
||||
}
|
||||
}
|
||||
|
|
@ -22,16 +22,12 @@ import (
|
|||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/compose-spec/compose-go/v2/types"
|
||||
"github.com/docker/compose/v2/internal/desktop"
|
||||
pathutil "github.com/docker/compose/v2/internal/paths"
|
||||
"github.com/docker/compose/v2/pkg/api"
|
||||
"github.com/docker/compose/v2/pkg/progress"
|
||||
"github.com/docker/compose/v2/pkg/prompt"
|
||||
|
|
@ -155,58 +151,6 @@ func (s *composeService) ensureProjectVolumes(ctx context.Context, project *type
|
|||
ids[k] = id
|
||||
}
|
||||
|
||||
err := func() error {
|
||||
if s.manageDesktopFileSharesEnabled(ctx) {
|
||||
// collect all the bind mount paths and try to set up file shares in
|
||||
// Docker Desktop for them
|
||||
var paths []string
|
||||
for _, svcName := range project.ServiceNames() {
|
||||
svc := project.Services[svcName]
|
||||
for _, vol := range svc.Volumes {
|
||||
if vol.Type != string(mount.TypeBind) {
|
||||
continue
|
||||
}
|
||||
p := filepath.Clean(vol.Source)
|
||||
if !filepath.IsAbs(p) {
|
||||
return fmt.Errorf("file share path is not absolute: %s", p)
|
||||
}
|
||||
if fi, err := os.Stat(p); errors.Is(err, fs.ErrNotExist) {
|
||||
// actual directory will be implicitly created when the
|
||||
// file share is initialized if it doesn't exist, so
|
||||
// need to filter out any that should not be auto-created
|
||||
if vol.Bind != nil && !vol.Bind.CreateHostPath {
|
||||
logrus.Debugf("Skipping creating file share for %q: does not exist and `create_host_path` is false", p)
|
||||
continue
|
||||
}
|
||||
} else if err != nil {
|
||||
// if we can't read the path, we won't be able to make
|
||||
// a file share for it
|
||||
logrus.Debugf("Skipping creating file share for %q: %v", p, err)
|
||||
continue
|
||||
} else if !fi.IsDir() {
|
||||
// ignore files & special types (e.g. Unix sockets)
|
||||
logrus.Debugf("Skipping creating file share for %q: not a directory", p)
|
||||
continue
|
||||
}
|
||||
|
||||
paths = append(paths, p)
|
||||
}
|
||||
}
|
||||
|
||||
// remove duplicate/unnecessary child paths and sort them for predictability
|
||||
paths = pathutil.EncompassingPaths(paths)
|
||||
sort.Strings(paths)
|
||||
|
||||
fileShareManager := desktop.NewFileShareManager(s.desktopCli, project.Name, paths)
|
||||
if err := fileShareManager.EnsureExists(ctx); err != nil {
|
||||
return fmt.Errorf("initializing file shares: %w", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}()
|
||||
if err != nil {
|
||||
progress.ContextWriter(ctx).TailMsgf("Failed to prepare Synchronized file shares: %v", err)
|
||||
}
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -17,11 +17,8 @@
|
|||
package compose
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/docker/compose/v2/internal/desktop"
|
||||
"github.com/docker/compose/v2/internal/experimental"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
func (s *composeService) SetDesktopClient(cli *desktop.Client) {
|
||||
|
|
@ -31,18 +28,3 @@ func (s *composeService) SetDesktopClient(cli *desktop.Client) {
|
|||
func (s *composeService) SetExperiments(experiments *experimental.State) {
|
||||
s.experiments = experiments
|
||||
}
|
||||
|
||||
func (s *composeService) manageDesktopFileSharesEnabled(ctx context.Context) bool {
|
||||
if !s.isDesktopIntegrationActive() {
|
||||
return false
|
||||
}
|
||||
|
||||
// synchronized file share support in Docker Desktop is dependent upon
|
||||
// a variety of factors (settings, OS, etc), which this endpoint abstracts
|
||||
fileSharesConfig, err := s.desktopCli.GetFileSharesConfig(ctx)
|
||||
if err != nil {
|
||||
logrus.Debugf("Failed to retrieve file shares config: %v", err)
|
||||
return false
|
||||
}
|
||||
return fileSharesConfig.Active && fileSharesConfig.Compose.ManageBindMounts
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,7 +23,6 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/compose-spec/compose-go/v2/types"
|
||||
"github.com/docker/compose/v2/internal/desktop"
|
||||
"github.com/docker/compose/v2/pkg/api"
|
||||
"github.com/docker/compose/v2/pkg/progress"
|
||||
"github.com/docker/compose/v2/pkg/utils"
|
||||
|
|
@ -157,13 +156,6 @@ func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.P
|
|||
})
|
||||
}
|
||||
|
||||
if s.manageDesktopFileSharesEnabled(ctx) {
|
||||
ops = append(ops, func() error {
|
||||
desktop.RemoveFileSharesForProject(ctx, s.desktopCli, project.Name)
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
return ops
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue