Initial distributor implementation

appengine/cmd/dm/model/execution.go

 // Copyright 2016 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package model
 
 import (
         "crypto/subtle"
         "encoding/hex"
         "errors"
         "fmt"
         "time"
 
         "google.golang.org/grpc/codes"
 
         "golang.org/x/net/context"
 
         "github.com/luci/gae/service/datastore"
-        "github.com/luci/luci-go/common/api/dm/service/v1"
+        dm "github.com/luci/luci-go/common/api/dm/service/v1"
+        "github.com/luci/luci-go/common/clock"
+        "github.com/luci/luci-go/common/cryptorand"
         "github.com/luci/luci-go/common/grpcutil"
         "github.com/luci/luci-go/common/logging"
         google_pb "github.com/luci/luci-go/common/proto/google"
 )
 
 const ek = logging.ErrorKey
 
 // Execution represents either an ongoing execution on the Quest's specified
 // distributor, or is a placeholder for an already-completed Execution.
 type Execution struct {
         ID      uint32         `gae:"$id"`
         Attempt *datastore.Key `gae:"$parent"`
 
-        State       dm.Execution_State
-        StateReason string `gae:",noindex"`
+        Created  time.Time
+        Modified time.Time
 
-        Created          time.Time
-        DistributorToken string
-        DistributorURL   string `gae:",noindex"`
+        DistributorConfigName    string
+        DistributorConfigVersion string
+        DistributorToken         string
+
+        State dm.Execution_State
+
+        // Only valid in the ABNORMAL_FINISHED state.
+        AbnormalFinish dm.AbnormalFinish
+
+        // Only valid in the FINISHED state.
+        ResultPersistentState []byte `gae:",noindex"`
+
+        // These are DM's internal mechanism for performing timeout actions on
+        // Executions.
+        //
+        // The TimeTo* variables are filled in by the distributor when this Execution
+        // is created.
+        //
+        // The Timeout is only active when the Execution is in a non-terminal state.
+        TimeToStart time.Duration `gae:",noindex"` // SCHEDULING -> RUNNING
+        TimeToRun   time.Duration `gae:",noindex"` // RUNNING -> STOPPING
+        TimeToStop  time.Duration `gae:",noindex"` // STOPPING -> FINISHED
 
         // Token is a randomized nonce that's used to verify that RPCs verify from the
         // expected client (the client that's currently running the Execution). The
         // Token has 2 modes.
         //
         // When the Execution is handed to the distributor, the Token is randomly
         // generated by DM and passed to the distributor. The State of the Execution
-        // starts as Scheduled. This token may be used by the client to "activate" the
+        // starts as SCHEDULED. This token may be used by the client to "activate" the
         // Execution with the ActivateExecution rpc. At that point, the client
-        // provides a new random token, the Execution State moves from Scheduled to
-        // Running, and Token assumes the new value. As long as the Execution State is
-        // running, the client may continue to use that new Token value to
+        // provides a new random token, the Execution State moves from SCHEDULED to
+        // RUNNING, and Token assumes the new value. As long as the Execution State is
+        // RUNNING, the client may continue to use that new Token value to
         // authenticate other rpc's like AddDeps and FinishAttempt.
         //
-        // As soon as the Execution is no longer supposed to have access, Token will
-        // be nil'd out.
+        // As soon as the Execution is in the STOPPING, ABNORMAL_FINISHED or FINISHED
+        // state, this will be nil'd out.
         Token []byte `gae:",noindex"`
 }
 
-// Revoke will null-out the Token and Put this Execution to the datastore.
+// MakeExecution makes a new Execution in the SCHEDULING state, with a new
+// random Token.
+func MakeExecution(c context.Context, e *dm.Execution_ID, cfgName, cfgVers string) *Execution {
+        now := clock.Now(c).UTC()
+        ret := &Execution{
+                ID:      e.Id,
+                Attempt: AttemptKeyFromID(c, e.AttemptID()),
+
+                Created:  now,
+                Modified: now,
+
+                DistributorConfigName:    cfgName,
+                DistributorConfigVersion: cfgVers,
+
+                Token: MakeRandomToken(c, dm.MinimumActivationTokenLength),
+        }
+        return ret
+}
+
+// ModifyState changes the current state of this Execution and updates its
+// Modified timestamp.
+func (e *Execution) ModifyState(c context.Context, newState dm.Execution_State) error {
+        if e.State == newState {
+                return nil
+        }
+        if err := e.State.Evolve(newState); err != nil {
+                return err
+        }
+        now := clock.Now(c).UTC()
+        if now.After(e.Modified) {
+                e.Modified = now
+        } else {
+                // Microsecond is the smallest granularity that datastore can store
+                // timestamps, so use that to disambiguate: the goal here is that any
+                // modification always increments the modified time, and never decrements
+                // it.
+                e.Modified = e.Modified.Add(time.Microsecond)
+        }
+        return nil
+}
+
+// MakeRandomToken creates a cryptographically random byte slice of the
+// specified length. It panics if the specified length cannot be read in full.
+func MakeRandomToken(c context.Context, l uint32) []byte {
+        rtok := make([]byte, l)
+        if _, err := cryptorand.Read(c, rtok); err != nil {
+                panic(err)
+        }
+        return rtok
+}
+
+// Revoke will clear the Token and Put this Execution to the datastore. This
+// action requires the Execution to be in the RUNNING state, and causes it to
+// enter the STOPPING state.
 func (e *Execution) Revoke(c context.Context) error {
         e.Token = nil
+        if err := e.ModifyState(c, dm.Execution_STOPPING); err != nil {
+                return err
+        }
         return datastore.Get(c).Put(e)
 }
 
 func loadExecution(c context.Context, eid *dm.Execution_ID) (a *Attempt, e *Execution, err error) {
         ds := datastore.Get(c)
 
         a = &Attempt{ID: *eid.AttemptID()}
         e = &Execution{ID: eid.Id, Attempt: ds.KeyForObj(a)}
         err = datastore.Get(c).Get(a, e)
 
         if err != nil {
-                err = fmt.Errorf("couldn't get attempt %v or its execution %d: %s", a.ID, e.ID, err)
+                err = grpcutil.Errf(codes.Internal,
+                        "couldn't get attempt %v or its execution %d: %s", a.ID, e.ID, err)
                 return
         }
 
         if a.CurExecution != e.ID {
                 err = fmt.Errorf("verifying incorrect execution %d, expected %d", a.CurExecution, e.ID)
                 return
         }
         return
 }
 
@@ skipping 12 matching lines @@
                 err = errors.New("Execution is not running")
                 return
         }
 
         if subtle.ConstantTimeCompare(e.Token, auth.Token) != 1 {
                 err = fmt.Errorf("incorrect Token: %x", hex.EncodeToString(auth.Token))
         }
         return
 }
 
+func makeError(err error, msg string) error {
+        code := grpcutil.Code(err)
+        if code == codes.Unknown {
+                code = codes.PermissionDenied
+        }
+        return grpcutil.Errf(code, msg)
+}
+
 // AuthenticateExecution verifies that the Attempt is executing, and that evkey
 // matches the execution key of the current Execution for this Attempt.
 //
 // As a bonus, it will return the loaded Attempt and Execution.
 func AuthenticateExecution(c context.Context, auth *dm.Execution_Auth) (a *Attempt, e *Execution, err error) {
         a, e, err = verifyExecutionAndCheckExTok(c, auth)
         if err != nil {
                 logging.Fields{ek: err, "eid": auth.Id}.Errorf(c, "failed to verify execution")
-                err = grpcutil.Errf(codes.Unauthenticated, "requires execution Auth")
+                err = makeError(err, "requires execution Auth")
         }
         return a, e, err
 }
 
 // InvalidateExecution verifies that the execution key is valid, and then
 // revokes the execution key.
 //
 // As a bonus, it will return the loaded Attempt and Execution.
 func InvalidateExecution(c context.Context, auth *dm.Execution_Auth) (a *Attempt, e *Execution, err error) {
         if a, e, err = verifyExecutionAndCheckExTok(c, auth); err != nil {
                 logging.Fields{ek: err, "eid": auth.Id}.Errorf(c, "failed to verify execution")
-                err = grpcutil.Errf(codes.Unauthenticated, "requires execution Auth")
+                err = makeError(err, "requires execution Auth")
                 return
         }
 
         err = e.Revoke(c)
         if err != nil {
                 logging.Fields{ek: err, "eid": auth.Id}.Errorf(c, "failed to revoke execution")
-                err = grpcutil.Errf(codes.Internal, "unable to invalidate Auth")
+                err = makeError(err, "unable to invalidate Auth")
         }
         return
 }
 
 func verifyExecutionAndActivate(c context.Context, auth *dm.Execution_Auth, actTok []byte) (a *Attempt, e *Execution, err error) {
         a, e, err = loadExecution(c, auth.Id)
         if err != nil {
                 return
         }
 
         if a.State != dm.Attempt_EXECUTING {
                 err = errors.New("Attempt is in wrong state")
                 return
         }
 
         switch e.State {
-        case dm.Execution_SCHEDULED:
+        case dm.Execution_SCHEDULING:
                 if subtle.ConstantTimeCompare(e.Token, auth.Token) != 1 {
                         err = errors.New("incorrect ActivationToken")
                         return
                 }
 
                 e.State.MustEvolve(dm.Execution_RUNNING)
                 e.Token = actTok
                 err = datastore.Get(c).Put(e)
+                logging.Infof(c, "activated execution %s: was SCHEDULING now RUNNING", auth.Id)
 
         case dm.Execution_RUNNING:
                 if subtle.ConstantTimeCompare(e.Token, actTok) != 1 {
                         err = errors.New("incorrect Token")
                 }
                 // either the Token matched, in which case this is simply a retry
                 // by the same client, so there's no error, or it's wrong which means it's
                 // a retry by a different client.
 
+                logging.Infof(c, "already activated execution %s", auth.Id)
+
         default:
                 err = fmt.Errorf("Execution is in wrong state")
         }
         return
 }
 
 // ActivateExecution validates that the execution is unactivated and that
 // the activation token matches and then sets the token to the new
 // value.
 //
 // It's OK to retry this. Subsequent invocations with the same Token
 // will recognize this case and not return an error.
 func ActivateExecution(c context.Context, auth *dm.Execution_Auth, actToken []byte) (a *Attempt, e *Execution, err error) {
         a, e, err = verifyExecutionAndActivate(c, auth, actToken)
         if err != nil {
                 logging.Fields{ek: err, "eid": auth.Id}.Errorf(c, "failed to activate execution")
-                err = grpcutil.Errf(codes.Unauthenticated, "failed to activate execution Auth")
+                err = makeError(err, "failed to activate execution Auth")
         }
         return a, e, err
 }
 
+// GetEID gets an Execution_ID for this Execution. It panics if the Execution
+// is in an invalid state.
+func (e *Execution) GetEID() *dm.Execution_ID {
+        aid := &dm.Attempt_ID{}
+        if e.ID == 0 {
+                panic("cannot create valid Execution_ID with 0-value ID field")
+        }
+        if err := aid.SetDMEncoded(e.Attempt.StringID()); err != nil {
+                panic(err)
+        }
+        return dm.NewExecutionID(aid.Quest, aid.Id, e.ID)
+}
+
 // ToProto returns a dm proto version of this Execution.
 func (e *Execution) ToProto(includeID bool) *dm.Execution {
-        ret := &dm.Execution{
-                Data: &dm.Execution_Data{
-                        State:              e.State,
-                        StateReason:        e.StateReason,
-                        Created:            google_pb.NewTimestamp(e.Created),
-                        DistributorToken:   e.DistributorToken,
-                        DistributorInfoUrl: e.DistributorURL,
-                },
-        }
+        ret := &dm.Execution{Data: e.DataProto()}
         if includeID {
-                aid := &dm.Attempt_ID{}
-                if err := aid.SetDMEncoded(e.Attempt.StringID()); err != nil {
-                        panic(err)
-                }
-                ret.Id = dm.NewExecutionID(aid.Quest, aid.Id, e.ID)
+                ret.Id = e.GetEID()
         }
         return ret
 }
+
+// DataProto returns an Execution.Data message for this Execution.
+//
+// This omits the DistributorInfo.Url portion, which must be filled in elsewhere for
+// package cyclical import reasons.
+func (e *Execution) DataProto() (ret *dm.Execution_Data) {
+        switch e.State {
+        case dm.Execution_SCHEDULING:
+                ret = dm.NewExecutionScheduling().Data
+        case dm.Execution_RUNNING:
+                ret = dm.NewExecutionRunning().Data
+        case dm.Execution_STOPPING:
+                ret = dm.NewExecutionStopping().Data
+        case dm.Execution_FINISHED:
+                ret = dm.NewExecutionFinished(string(e.ResultPersistentState)).Data
+        case dm.Execution_ABNORMAL_FINISHED:
+                ret = dm.NewExecutionAbnormalFinish(&e.AbnormalFinish).Data
+        default:
+                panic(fmt.Errorf("unknown Execution_State: %s", e.State))
+        }
+        ret.Created = google_pb.NewTimestamp(e.Created)
+        ret.Modified = google_pb.NewTimestamp(e.Modified)
+        ret.DistributorInfo = &dm.Execution_Data_DistributorInfo{
+                ConfigName:    e.DistributorConfigName,
+                ConfigVersion: e.DistributorConfigVersion,
+                Token:         e.DistributorToken,
+        }
+        return ret
+}