WIP: Remove state from operations.

Signed-off-by: Mark Laing <mark.laing@canonical.com>

lxd-agent/exec.go

@@ -136,7 +136,8 @@ func execPost(d *Daemon, r *http.Request) response.Response {
 
  resources := map[string][]api.URL{}
 
- op, err := operations.OperationCreate(nil, "", operations.OperationClassWebsocket, operationtype.CommandExec, resources, ws.Metadata(), ws.Do, nil, ws.Connect, r)
+ operationOpts := operations.Options().WithResources(resources).WithMetadata(ws.Metadata()).WithOnConnect(ws.Connect).WithRequest(r)
+ op, err := operations.OperationCreate(context.Background(), operations.OperationClassWebsocket, operationtype.CommandExec, "lxd-agent", d.events, ws.Do, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }
@@ -163,6 +164,7 @@ type execWs struct {
  cwd string
 }
 
+// Metadata returns information required for the client to interact with the exec websocket.
 func (s *execWs) Metadata() any {
  fds := shared.Jmap{}
  for fd, secret := range s.fds {
@@ -181,6 +183,7 @@ func (s *execWs) Metadata() any {
  }
 }
 
+// Connect is the websocket connect hook for the exec operation.
 func (s *execWs) Connect(op *operations.Operation, r *http.Request, w http.ResponseWriter) error {
  secret := r.FormValue("secret")
  if secret == "" {
@@ -211,9 +214,9 @@ func (s *execWs) Connect(op *operations.Operation, r *http.Request, w http.Respo
  return nil
  } else if !found {
  return fmt.Errorf("Unknown websocket number")
- } else {
- return fmt.Errorf("Websocket number already connected")
  }
+
+ return fmt.Errorf("Websocket number already connected")
  }
  }
 
@@ -222,6 +225,7 @@ func (s *execWs) Connect(op *operations.Operation, r *http.Request, w http.Respo
  return os.ErrPermission
 }
 
+// Do runs waits for the websocket to connect, then proxies the VM ttys through the websocket.
 func (s *execWs) Do(op *operations.Operation) error {
  // Once this function ends ensure that any connected websockets are closed.
  defer func() {
@@ -239,7 +243,6 @@ func (s *execWs) Do(op *operations.Operation) error {
  logger.Debug("Waiting for exec websockets to connect")
  select {
  case <-s.requiredConnectedCtx.Done():
- break
  case <-time.After(time.Second * 5):
  return fmt.Errorf("Timed out waiting for websockets to connect")
  }

lxd/acme.go

@@ -137,7 +137,7 @@ func autoRenewCertificate(ctx context.Context, d *Daemon, force bool) error {
  return nil
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.RenewServerCertificate, nil, nil, opRun, nil, nil, nil)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.RenewServerCertificate, s.ServerName, s.Events, opRun, operations.ClusterOptions(s.DB.Cluster.TransactionSQL))
  if err != nil {
  logger.Error("Failed creating renew server certificate operation", logger.Ctx{"err": err})
  return err

lxd/api_cluster.go

@@ -461,7 +461,8 @@ func clusterPutBootstrap(d *Daemon, r *http.Request, req api.ClusterPut) respons
  d.localConfig = config
  d.globalConfigMu.Unlock()
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.ClusterBootstrap, resources, nil, run, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithResources(resources).WithRequest(r)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ClusterBootstrap, s.ServerName, s.Events, run, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }
@@ -869,7 +870,8 @@ func clusterPutJoin(d *Daemon, r *http.Request, req api.ClusterPut) response.Res
  resources := map[string][]api.URL{}
  resources["cluster"] = []api.URL{}
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.ClusterJoin, resources, nil, run, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithResources(resources).WithRequest(r)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ClusterJoin, s.ServerName, s.Events, run, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }
@@ -1433,7 +1435,8 @@ func clusterNodesPost(d *Daemon, r *http.Request) response.Response {
  resources := map[string][]api.URL{}
  resources["cluster"] = []api.URL{}
 
- op, err := operations.OperationCreate(s, api.ProjectDefaultName, operations.OperationClassToken, operationtype.ClusterJoinToken, resources, meta, nil, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithProjectName(api.ProjectDefaultName).WithResources(resources).WithMetadata(meta).WithRequest(r)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassToken, operationtype.ClusterJoinToken, s.ServerName, s.Events, nil, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }
@@ -3276,7 +3279,8 @@ func evacuateClusterMember(s *state.State, gateway *cluster.Gateway, r *http.Req
  return nil
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.ClusterMemberEvacuate, nil, nil, run, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithRequest(r)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ClusterMemberEvacuate, s.ServerName, s.Events, run, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }
@@ -3601,7 +3605,8 @@ func restoreClusterMember(d *Daemon, r *http.Request) response.Response {
  return nil
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.ClusterMemberRestore, nil, nil, run, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithRequest(r)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ClusterMemberRestore, s.ServerName, s.Events, run, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }
@@ -4492,7 +4497,7 @@ func autoHealClusterTask(d *Daemon) (task.Func, task.Schedule) {
  return nil
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.ClusterHeal, nil, nil, opRun, nil, nil, nil)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ClusterHeal, s.ServerName, s.Events, opRun, operations.ClusterOptions(s.DB.Cluster.TransactionSQL))
  if err != nil {
  logger.Error("Failed creating cluster instances heal operation", logger.Ctx{"err": err})
  return

lxd/api_project.go

@@ -819,7 +819,8 @@ func projectPost(d *Daemon, r *http.Request) response.Response {
  return nil
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.ProjectRename, nil, nil, run, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithRequest(r)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ProjectRename, s.ServerName, s.Events, run, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }

lxd/backup.go

@@ -314,7 +314,7 @@ func pruneExpiredBackupsTask(d *Daemon) (task.Func, task.Schedule) {
  return nil
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.BackupsExpire, nil, nil, opRun, nil, nil, nil)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.BackupsExpire, s.ServerName, s.Events, opRun, operations.ClusterOptions(s.DB.Cluster.TransactionSQL))
  if err != nil {
  logger.Error("Failed creating expired backups operation", logger.Ctx{"err": err})
  return

lxd/certificates.go

@@ -596,7 +596,8 @@ func certificatesPost(d *Daemon, r *http.Request) response.Response {
  meta["expiresAt"] = expiresAt
  }
 
- op, err := operations.OperationCreate(s, api.ProjectDefaultName, operations.OperationClassToken, operationtype.CertificateAddToken, nil, meta, nil, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithRequest(r).WithMetadata(meta).WithProjectName(api.ProjectDefaultName)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassToken, operationtype.CertificateAddToken, s.ServerName, s.Events, nil, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }

lxd/db/db.go

@@ -332,6 +332,13 @@ func (c *Cluster) Transaction(ctx context.Context, f func(context.Context, *Clus
  return c.transaction(ctx, f)
 }
 
+// TransactionSQL is identical to Transaction but the hook has a sql.Tx instead (for import cycle reasons).
+func (c *Cluster) TransactionSQL(ctx context.Context, f func(context.Context, *sql.Tx) error) error {
+ c.mu.RLock()
+ defer c.mu.RUnlock()
+ return c.transactionSQL(ctx, f)
+}
+
 // EnterExclusive acquires a lock on the cluster db, so any successive call to
 // Transaction will block until ExitExclusive has been called.
 func (c *Cluster) EnterExclusive() error {
@@ -384,6 +391,21 @@ func (c *Cluster) transaction(ctx context.Context, f func(context.Context, *Clus
  })
 }
 
+func (c *Cluster) transactionSQL(ctx context.Context, f func(context.Context, *sql.Tx) error) error {
+ return query.Retry(ctx, func(ctx context.Context) error {
+ err := query.Transaction(ctx, c.db, f)
+ if errors.Is(err, context.DeadlineExceeded) {
+ // If the query timed out it likely means that the leader has abruptly become unreachable.
+ // Now that this query has been cancelled, a leader election should have taken place by now.
+ // So let's retry the transaction once more in case the global database is now available again.
+ logger.Warn("Transaction timed out. Retrying once", logger.Ctx{"member": c.nodeID, "err": err})
+ return query.Transaction(ctx, c.db, f)
+ }
+
+ return err
+ })
+}
+
 // NodeID sets the node NodeID associated with this cluster instance. It's used for
 // backward-compatibility of all db-related APIs that were written before
 // clustering and don't accept a node NodeID, so in those cases we automatically
@@ -432,7 +454,7 @@ func begin(db *sql.DB) (*sql.Tx, error) {
  }
 
  logger.Debugf("DbBegin: DB still locked")
- logger.Debugf(logger.GetStack())
+ logger.Debug(logger.GetStack())
  return nil, fmt.Errorf("DB is locked")
 }
 

lxd/images.go

@@ -1322,7 +1322,8 @@ func imagesPost(d *Daemon, r *http.Request) response.Response {
  }
  }
 
- imageOp, err := operations.OperationCreate(s, projectName, operations.OperationClassTask, operationtype.ImageDownload, nil, metadata, run, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithProjectName(projectName).WithMetadata(metadata).WithRequest(r)
+ imageOp, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ImageDownload, s.ServerName, s.Events, run, operationOpts)
  if err != nil {
  cleanup(builddir, post)
  return response.InternalError(err)
@@ -1763,7 +1764,8 @@ func autoUpdateImagesTask(d *Daemon) (task.Func, task.Schedule) {
  return autoUpdateImages(ctx, s)
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.ImagesUpdate, nil, nil, opRun, nil, nil, nil)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ImagesUpdate, s.ServerName, s.Events, opRun, operationOpts)
  if err != nil {
  logger.Error("Failed creating image update operation", logger.Ctx{"err": err})
  return
@@ -2412,7 +2414,8 @@ func pruneExpiredImagesTask(d *Daemon) (task.Func, task.Schedule) {
  return pruneExpiredImages(ctx, s, op)
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.ImagesExpire, nil, nil, opRun, nil, nil, nil)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ImagesExpire, s.ServerName, s.Events, opRun, operationOpts)
  if err != nil {
  logger.Error("Failed creating expired image prune operation", logger.Ctx{"err": err})
  return
@@ -2530,7 +2533,8 @@ func pruneLeftoverImages(s *state.State) {
  return nil
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.ImagesPruneLeftover, nil, nil, opRun, nil, nil, nil)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ImagesPruneLeftover, s.ServerName, s.Events, opRun, operationOpts)
  if err != nil {
  logger.Error("Failed creating leftover image clean up operation", logger.Ctx{"err": err})
  return
@@ -2889,7 +2893,8 @@ func imageDelete(d *Daemon, r *http.Request) response.Response {
  resources := map[string][]api.URL{}
  resources["images"] = []api.URL{*api.NewURL().Path(version.APIVersion, "images", details.image.Fingerprint)}
 
- op, err := operations.OperationCreate(s, projectName, operations.OperationClassTask, operationtype.ImageDelete, resources, nil, do, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithProjectName(projectName).WithResources(resources).WithRequest(r)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ImageDelete, s.ServerName, s.Events, do, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }
@@ -4400,7 +4405,8 @@ func imageExportPost(d *Daemon, r *http.Request) response.Response {
  return nil
  }
 
- op, err := operations.OperationCreate(s, projectName, operations.OperationClassTask, operationtype.ImageDownload, nil, nil, run, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithProjectName(projectName).WithRequest(r)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ImageDownload, s.ServerName, s.Events, run, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }
@@ -4587,7 +4593,8 @@ func imageRefresh(d *Daemon, r *http.Request) response.Response {
  return err
  }
 
- op, err := operations.OperationCreate(s, projectName, operations.OperationClassTask, operationtype.ImageRefresh, nil, nil, run, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithProjectName(projectName).WithRequest(r)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ImageRefresh, s.ServerName, s.Events, run, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }
@@ -4622,7 +4629,8 @@ func autoSyncImagesTask(d *Daemon) (task.Func, task.Schedule) {
  return autoSyncImages(ctx, s)
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.ImagesSynchronize, nil, nil, opRun, nil, nil, nil)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.ImagesSynchronize, s.ServerName, s.Events, opRun, operationOpts)
  if err != nil {
  logger.Error("Failed creating image synchronization operation", logger.Ctx{"err": err})
  return
@@ -4825,7 +4833,8 @@ func createTokenResponse(s *state.State, r *http.Request, projectName string, fi
  resources := map[string][]api.URL{}
  resources["images"] = []api.URL{*api.NewURL().Path(version.APIVersion, "images", fingerprint)}
 
- op, err := operations.OperationCreate(s, projectName, operations.OperationClassToken, operationtype.ImageToken, resources, meta, nil, nil, nil, r)
+ operationOpts := operations.ClusterOptions(s.DB.Cluster.TransactionSQL).WithProjectName(projectName).WithResources(resources).WithMetadata(meta).WithRequest(r)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassToken, operationtype.ImageToken, s.ServerName, s.Events, nil, operationOpts)
  if err != nil {
  return response.InternalError(err)
  }

lxd/instance.go

@@ -664,7 +664,7 @@ func pruneExpiredAndAutoCreateInstanceSnapshotsTask(d *Daemon) (task.Func, task.
  return pruneExpiredInstanceSnapshots(ctx, expiredSnapshotInstances)
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.SnapshotsExpire, nil, nil, opRun, nil, nil, nil)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.SnapshotsExpire, s.ServerName, s.Events, opRun, operations.ClusterOptions(s.DB.Cluster.TransactionSQL))
  if err != nil {
  logger.Error("Failed creating instance snapshots expiry operation", logger.Ctx{"err": err})
  } else {
@@ -690,7 +690,7 @@ func pruneExpiredAndAutoCreateInstanceSnapshotsTask(d *Daemon) (task.Func, task.
  return autoCreateInstanceSnapshots(ctx, s, instances)
  }
 
- op, err := operations.OperationCreate(s, "", operations.OperationClassTask, operationtype.SnapshotCreate, nil, nil, opRun, nil, nil, nil)
+ op, err := operations.OperationCreate(s.ShutdownCtx, operations.OperationClassTask, operationtype.SnapshotCreate, s.ServerName, s.Events, opRun, operations.ClusterOptions(s.DB.Cluster.TransactionSQL))
  if err != nil {
  logger.Error("Failed creating scheduled instance snapshot operation", logger.Ctx{"err": err})
  } else {

lxd/instance/drivers/driver_lxc.go

@@ -5467,12 +5467,11 @@ func (d *lxc) MigrateSend(args instance.MigrateSendArgs) error {
  }
 
  actionScriptOp, err := operations.OperationCreate(
- d.state,
- d.Project().Name,
+ d.state.ShutdownCtx,
  operations.OperationClassWebsocket,
  operationtype.InstanceLiveMigrate,
- nil,
- nil,
+ d.state.ServerName,
+ d.state.Events,
  func(op *operations.Operation) error {
  result := <-restoreSuccess
  if !result {
@@ -5481,28 +5480,28 @@ func (d *lxc) MigrateSend(args instance.MigrateSendArgs) error {
 
  return nil
  },
- nil,
- func(op *operations.Operation, r *http.Request, w http.ResponseWriter) error {
- secret := r.FormValue("secret")
- if secret == "" {
- return fmt.Errorf("Missing action script secret")
- }
+ operations.ClusterOptions(d.state.DB.Cluster.TransactionSQL).
+ WithProjectName(d.Project().Name).
+ WithOnConnect(func(op *operations.Operation, r *http.Request, w http.ResponseWriter) error {
+ secret := r.FormValue("secret")
+ if secret == "" {
+ return fmt.Errorf("Missing action script secret")
+ }
 
- if secret != actionScriptOpSecret {
- return os.ErrPermission
- }
+  if secret != actionScriptOpSecret {
+  return os.ErrPermission
+  }
 
- c, err := ws.Upgrader.Upgrade(w, r, nil)
- if err != nil {
- return err
- }
+  c, err := ws.Upgrader.Upgrade(w, r, nil)
+  if err != nil {
+  return err
+  }
 
- dumpDone <- true
+  dumpDone <- true
 
- closeMsg := websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")
- return c.WriteMessage(websocket.CloseMessage, closeMsg)
- },
- nil,
+ closeMsg := websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")
+ return c.WriteMessage(websocket.CloseMessage, closeMsg)
+ }),
  )
  if err != nil {
  _ = os.RemoveAll(checkpointDir)