iclient/incus_operation.go

Rama develop · réplica de lectura del repositorio original.

package iclient

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"time"

	"github.com/lxc/incus/v7/shared/api"
)

// incusOperationsPath is the collection every operation call hangs off.
const incusOperationsPath = "/operations"

// incusOperationBuffer holds updates for a caller that is between reads.
const incusOperationBuffer = 8

// incusOperationRetry is how long an operation keeps trying to open or regain
// its event socket before it gives up on the server.
const incusOperationRetry = 30 * time.Second

// incusOperationRetryDelay is the pause between those attempts.
const incusOperationRetryDelay = 2 * time.Second

// asyncOperation issues a request the server answers asynchronously and
// returns the operation's updates.
//
// The listener opens before the request goes out, so an operation that finishes
// immediately is still reported.
//
// The channel carries the operation as accepted, then every update, and closes
// on a terminal state - the last value is the outcome, including its Err.
//
// A token operation is the exception: it waits to be used rather than
// finishing, so read its first value and cancel the context.
func (c *Connection) asyncOperation(ctx context.Context, project string, method string, path string, query url.Values, body any, etag string) (<-chan api.Operation, error) {
	return c.async(ctx, project, method+" "+path, func(sendCtx context.Context) (*api.Response, error) {
		resp, _, err := c.do(sendCtx, project, method, path, query, body, etag)

		return resp, err
	})
}

// asyncUpload is asyncOperation for a request that streams a body, which is
// how an image is imported from tarballs. header carries what cannot be JSON.
func (c *Connection) asyncUpload(ctx context.Context, project string, path string, body io.Reader, contentType string, header http.Header) (<-chan api.Operation, error) {
	return c.async(ctx, project, http.MethodPost+" "+path, func(sendCtx context.Context) (*api.Response, error) {
		resp, _, err := c.send(sendCtx, http.MethodPost, uriFor(project, path, nil), body, contentType, "", header)

		return resp, err
	})
}

// async subscribes, then sends, then follows whatever operation came back.
//
// project has to be the one the operation runs in: incusd filters the event
// stream by project, so a listener anywhere else never sees the updates and the
// caller blocks until its context ends.
func (c *Connection) async(ctx context.Context, project string, what string, send func(context.Context) (*api.Response, error)) (<-chan api.Operation, error) {
	listenCtx, cancel := context.WithCancel(ctx)

	events, err := c.listenOperationEvents(listenCtx, project)
	if err != nil {
		cancel()

		return nil, err
	}

	resp, err := send(ctx)
	if err != nil {
		cancel()

		return nil, err
	}

	if resp.Type != api.AsyncResponse {
		cancel()

		return nil, fmt.Errorf("%s: expected an async response, got %q", what, resp.Type)
	}

	started := api.Operation{}

	err = resp.MetadataAsStruct(&started)
	if err != nil {
		cancel()

		return nil, fmt.Errorf("decoding the operation of %s: %w", what, err)
	}

	return c.followOperation(listenCtx, cancel, project, events, started), nil
}

// listenOperationEvents opens the operation listener, trying again while the
// server is too slow to accept it. Nothing has been sent yet, so a retry cannot
// run anything twice.
func (c *Connection) listenOperationEvents(ctx context.Context, project string) (<-chan api.Event, error) {
	deadline := time.Now().Add(incusOperationRetry)

	for {
		events, err := c.ListenEvents(ctx, project, []string{api.EventTypeOperation})
		if err == nil {
			return events, nil
		}

		// Only a timeout is worth another try, a server that answered will answer the same again.
		var netErr net.Error
		if !errors.As(err, &netErr) || !netErr.Timeout() {
			return nil, err
		}

		if ctx.Err() != nil || !time.Now().Before(deadline) {
			return nil, err
		}

		select {
		case <-time.After(incusOperationRetryDelay):
		case <-ctx.Done():
			return nil, err
		}
	}
}

// followOperation feeds the caller's channel: the operation as it stands, then
// every update the event stream carries for it, until it reaches an end state.
//
// A dead event socket is not a dead operation, so a stream that ends first is
// waited out on the wait endpoint rather than reported as a failure.
func (c *Connection) followOperation(ctx context.Context, cancel context.CancelFunc, project string, events <-chan api.Event, started api.Operation) <-chan api.Operation {
	updates := make(chan api.Operation, incusOperationBuffer)

	go func() {
		defer close(updates)

		// Stops the event listener, whichever way this returns.
		defer cancel()

		if !emitOperation(ctx, updates, started) {
			return
		}

		current := started

		for event := range events {
			update := api.Operation{}

			err := json.Unmarshal(event.Metadata, &update)
			if err != nil || update.ID != started.ID {
				continue
			}

			current = update

			if !emitOperation(ctx, updates, update) {
				return
			}
		}

		// The caller let go, so there is nobody left to report the outcome to.
		if ctx.Err() != nil {
			return
		}

		final, err := c.waitOutOperation(ctx, project, started.ID)
		if err != nil {
			// Say what is known: the operation was lost, not that it failed.
			current.StatusCode = api.Failure
			current.Status = api.Failure.String()
			current.Err = fmt.Sprintf("lost the event socket and could not reach the operation: %v", err)

			emitOperation(ctx, updates, current)

			return
		}

		emitOperation(ctx, updates, *final)
	}()

	return updates
}

// waitOutOperation picks an operation back up on the wait endpoint, which the
// server holds open and so costs no socket of its own. It keeps trying while
// the server is unreachable and gives up after incusOperationRetry of that.
func (c *Connection) waitOutOperation(ctx context.Context, project string, id string) (*api.Operation, error) {
	deadline := time.Now().Add(incusOperationRetry)

	for {
		op, err := c.WaitOperationID(ctx, project, id)
		if err == nil {
			return op, nil
		}

		// The server answered and does not have it, so there is nothing left to wait for.
		if api.StatusErrorCheck(err, http.StatusNotFound) {
			return nil, err
		}

		if ctx.Err() != nil || !time.Now().Before(deadline) {
			return nil, err
		}

		select {
		case <-time.After(incusOperationRetryDelay):
		case <-ctx.Done():
			return nil, err
		}
	}
}

// emitOperation delivers one update and reports whether to keep going.
func emitOperation(ctx context.Context, updates chan<- api.Operation, op api.Operation) bool {
	select {
	case updates <- op:
	case <-ctx.Done():
		return false
	}

	return !op.StatusCode.IsFinal()
}

// GetOperations returns the operations running in project.
func (c *Connection) GetOperations(ctx context.Context, project string) ([]api.Operation, error) {
	// Grouped by status, e.g. {"running": [...]}.
	byStatus := map[string][]api.Operation{}

	query := url.Values{}
	query.Set("recursion", "1")

	_, err := c.getStruct(ctx, project, incusOperationsPath, query, &byStatus)
	if err != nil {
		return nil, err
	}

	operations := []api.Operation{}
	for _, group := range byStatus {
		operations = append(operations, group...)
	}

	return operations, nil
}

// ListenOperation follows an operation this connection did not start. It
// cannot subscribe before the operation exists, so it reads the operation once
// after subscribing.
func (c *Connection) ListenOperation(ctx context.Context, project string, op api.Operation) (<-chan api.Operation, error) {
	listenCtx, cancel := context.WithCancel(ctx)

	events, err := c.ListenEvents(listenCtx, project, []string{api.EventTypeOperation})
	if err != nil {
		cancel()

		return nil, err
	}

	current := api.Operation{}

	_, err = c.getStruct(ctx, project, incusOperationsPath+"/"+url.PathEscape(op.ID), nil, &current)
	if err != nil {
		cancel()

		return nil, err
	}

	return c.followOperation(listenCtx, cancel, project, events, current), nil
}

// WaitOperationID blocks until the operation ends and returns how it ended.
//
// The server holds the request open, so unlike ListenOperation this costs one
// request rather than an event socket.
func (c *Connection) WaitOperationID(ctx context.Context, project string, id string) (*api.Operation, error) {
	operation := api.Operation{}

	// -1 stops the server applying a timeout of its own and answering early.
	query := url.Values{}
	query.Set("timeout", "-1")

	_, err := c.getStruct(ctx, project, incusOperationsPath+"/"+url.PathEscape(id)+"/wait", query, &operation)
	if err != nil {
		return nil, err
	}

	return &operation, nil
}

// CancelOperation asks the server to cancel an operation.
func (c *Connection) CancelOperation(ctx context.Context, project string, op api.Operation) error {
	_, _, err := c.do(ctx, project, http.MethodDelete, incusOperationsPath+"/"+url.PathEscape(op.ID), nil, nil, "")

	return err
}

// WaitOperation reads an operation to its end and returns the outcome. The
// last value is the result, so a failure comes back as an error here.
//
// Never call it on a token operation, which waits to be used rather than finishing.
func WaitOperation(ctx context.Context, updates <-chan api.Operation) (api.Operation, error) {
	last := api.Operation{}

	for {
		select {
		case update, ok := <-updates:
			if !ok {
				if last.Err != "" {
					// Incus takes the instance lock in the driver, so a busy write fails from here.
					return last, busyError(
						fmt.Errorf("operation %s: %s", last.ID, last.Err), last.Err)
				}

				if !last.StatusCode.IsFinal() {
					return last, fmt.Errorf("operation %s ended at %q", last.ID, last.Status)
				}

				return last, nil
			}

			last = update

		case <-ctx.Done():
			return last, ctx.Err()
		}
	}
}