Implement computeplan events (#98)

* Implement computeplan events

* Initialize event struct

* Add Cancel situation and renaming

* Rename after review

Author: thibaultrobert

chaincode/compute_plan.go

@@ -296,6 +296,7 @@ func cancelComputePlan(db *LedgerDB, args []string) (resp outputComputePlan, err
 		return outputComputePlan{}, err
 	}
 
+	db.AddComputePlanEvent(inp.Key, computeplan.State.Status)
 	resp.Fill(inp.Key, computeplan)
 	return resp, nil
 }
@@ -334,10 +335,10 @@ func (cp *ComputePlan) SaveState(db *LedgerDB) error {
 	return db.Put(cp.StateKey, cp.State)
 }
 
-// CheckNewTupleStatus check the tuple status (from an updated tuple or a new one)
+// UpdateStatus check the tuple status (from an updated tuple or a new one)
 // and, if required, it updates the compute plan' status and/or its doneCount.
 // It returns true if there is any change to the compute plan, false otherwise.
-func (cp *ComputePlan) CheckNewTupleStatus(tupleStatus string) bool {
+func (cp *ComputePlan) UpdateStatus(tupleStatus string) bool {
 	switch cp.State.Status {
 	case StatusFailed, StatusCanceled:
 	case StatusDone:
@@ -388,7 +389,7 @@ func (cp *ComputePlan) AddTuple(tupleType AssetType, key, status string) {
 		cp.TesttupleKeys = append(cp.TesttupleKeys, key)
 	}
 	cp.State.TupleCount++
-	cp.CheckNewTupleStatus(status)
+	cp.UpdateStatus(status)
 }
 
 // UpdateComputePlanState retreive the compute plan if the ID is not empty,
@@ -401,7 +402,8 @@ func UpdateComputePlanState(db *LedgerDB, ComputePlanID, tupleStatus, tupleKey s
 	if err != nil {
 		return err
 	}
-	if cp.CheckNewTupleStatus(tupleStatus) {
+	if cp.UpdateStatus(tupleStatus) {
+		db.AddComputePlanEvent(ComputePlanID, cp.State.Status)
 		return cp.SaveState(db)
 	}
 	return nil

chaincode/compute_plan_test.go

@@ -164,8 +164,8 @@ func TestModelCompositionComputePlanWorkflow(t *testing.T) {
 
 	out, err := createComputePlanInternal(db, modelCompositionComputePlan, tag)
 	assert.NoError(t, err)
-	assert.NotNil(t, db.tuplesEvent)
-	assert.Len(t, db.tuplesEvent.CompositeTraintuples, 2)
+	assert.NotNil(t, db.event)
+	assert.Len(t, db.event.CompositeTraintuples, 2)
 
 	// ensure the returned ranks are correct
 	validateTupleRank(t, db, 0, out.CompositeTraintupleKeys[0], CompositeTraintupleType)
@@ -179,7 +179,7 @@ func TestModelCompositionComputePlanWorkflow(t *testing.T) {
 	_, err = logStartCompositeTrain(db, assetToArgs(inputKey{out.CompositeTraintupleKeys[1]}))
 	assert.NoError(t, err)
 
-	db.tuplesEvent = &TuplesEvent{}
+	db.event = &Event{}
 	inpLogCompo := inputLogSuccessCompositeTrain{}
 	inpLogCompo.fillDefaults()
 	inpLogCompo.Key = out.CompositeTraintupleKeys[0]
@@ -189,12 +189,12 @@ func TestModelCompositionComputePlanWorkflow(t *testing.T) {
 	inpLogCompo.Key = out.CompositeTraintupleKeys[1]
 	_, err = logSuccessCompositeTrain(db, assetToArgs(inpLogCompo))
 	assert.NoError(t, err)
-	assert.Len(t, db.tuplesEvent.Testtuples, 2)
-	for _, test := range db.tuplesEvent.Testtuples {
+	assert.Len(t, db.event.Testtuples, 2)
+	for _, test := range db.event.Testtuples {
 		assert.Equalf(t, StatusTodo, test.Status, "blame it on %+v", test)
 	}
-	require.Len(t, db.tuplesEvent.Aggregatetuples, 1)
-	assert.Equal(t, StatusTodo, db.tuplesEvent.Aggregatetuples[0].Status)
+	require.Len(t, db.event.Aggregatetuples, 1)
+	assert.Equal(t, StatusTodo, db.event.Aggregatetuples[0].Status)
 
 	_, err = logStartAggregate(db, assetToArgs(inputKey{out.AggregatetupleKeys[0]}))
 	assert.NoError(t, err)
@@ -211,16 +211,16 @@ func TestModelCompositionComputePlanWorkflow(t *testing.T) {
 	_, err = logStartCompositeTrain(db, assetToArgs(inputKey{out.CompositeTraintupleKeys[3]}))
 	assert.NoError(t, err)
 
-	db.tuplesEvent = &TuplesEvent{}
+	db.event = &Event{}
 	inpLogCompo.Key = out.CompositeTraintupleKeys[2]
 	_, err = logSuccessCompositeTrain(db, assetToArgs(inpLogCompo))
 	assert.NoError(t, err)
 
 	inpLogCompo.Key = out.CompositeTraintupleKeys[3]
 	_, err = logSuccessCompositeTrain(db, assetToArgs(inpLogCompo))
 	assert.NoError(t, err)
-	assert.Len(t, db.tuplesEvent.Testtuples, 2)
-	for _, test := range db.tuplesEvent.Testtuples {
+	assert.Len(t, db.event.Testtuples, 2)
+	for _, test := range db.event.Testtuples {
 		assert.Equalf(t, StatusTodo, test.Status, "blame it on %+v", test)
 	}
 }
@@ -487,8 +487,8 @@ func TestCancelComputePlan(t *testing.T) {
 
 	out, err := createComputePlanInternal(db, modelCompositionComputePlan, tag)
 	assert.NoError(t, err)
-	assert.NotNil(t, db.tuplesEvent)
-	assert.Len(t, db.tuplesEvent.CompositeTraintuples, 2)
+	assert.NotNil(t, db.event)
+	assert.Len(t, db.event.CompositeTraintuples, 2)
 
 	_, err = cancelComputePlan(db, assetToArgs(inputKey{Key: out.ComputePlanID}))
 	assert.NoError(t, err)

chaincode/go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/hashicorp/go-version v1.2.0 // indirect
 	github.com/hyperledger/fabric v1.4.1
 	github.com/hyperledger/fabric-amcl v0.0.0-20181230093703-5ccba6eab8d6 // indirect
+	github.com/karrick/godirwalk v1.15.3 // indirect
 	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
 	github.com/kr/pretty v0.1.0 // indirect
 	github.com/leodido/go-urn v1.1.0 // indirect
@@ -35,6 +36,8 @@ require (
 	github.com/stretchr/objx v0.2.0 // indirect
 	github.com/stretchr/testify v1.3.0
 	github.com/sykesm/zap-logfmt v0.0.2 // indirect
+	github.com/uudashr/gopkgs v2.0.1+incompatible // indirect
+	github.com/uudashr/gopkgs/v2 v2.1.2 // indirect
 	go.uber.org/atomic v1.4.0 // indirect
 	go.uber.org/zap v1.10.0 // indirect
 	golang.org/x/crypto v0.0.0-20190513172903-22d7a77e9e5f // indirect

chaincode/go.sum

@@ -84,6 +84,9 @@ github.com/hyperledger/fabric-amcl v0.0.0-20181230093703-5ccba6eab8d6 h1:URjjUy3
 github.com/hyperledger/fabric-amcl v0.0.0-20181230093703-5ccba6eab8d6/go.mod h1:X+DIyUsaTmalOpmpQfIvFZjKHQedrURQ5t4YqquX7lE=
 github.com/ijc/Gotty v0.0.0-20170406111628-a8b993ba6abd h1:anPrsicrIi2ColgWTVPk+TrN42hJIWlfPHSBP9S0ZkM=
 github.com/ijc/Gotty v0.0.0-20170406111628-a8b993ba6abd/go.mod h1:3LVOLeyx9XVvwPgrt2be44XgSqndprz1G18rSk8KD84=
+github.com/karrick/godirwalk v1.12.0/go.mod h1:H5KPZjojv4lE+QYImBI8xVtrBRgYrIVsaRPx4tDPEn4=
+github.com/karrick/godirwalk v1.15.3 h1:0a2pXOgtB16CqIqXTiT7+K9L73f74n/aNQUnH6Ortew=
+github.com/karrick/godirwalk v1.15.3/go.mod h1:j4mkqPuvaLI8mp1DroR3P6ad7cyYd4c1qeJ3RV7ULlk=
 github.com/kisielk/errcheck v1.1.0 h1:ZqfnKyx9KGpRcW04j5nnPDgRgoXUeLh2YFBeFzphcA0=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/gotool v1.0.0 h1:AV2c/EiW3KqPNT9ZKl07ehoAGi4C5/01Cfbblndcapg=
@@ -163,6 +166,10 @@ github.com/sykesm/zap-logfmt v0.0.2 h1:czSzn+PIXCOAP/4NAIHTTziIKB8201PzoDkKTn+VR
 github.com/sykesm/zap-logfmt v0.0.2/go.mod h1:TerDJT124HaO8UTpZ2wJCipJRAKQ9XONM1mzUabIh6M=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8 h1:3SVOIvH7Ae1KRYyQWRjXWJEA9sS/c/pjvH++55Gr648=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
+github.com/uudashr/gopkgs v2.0.1+incompatible h1:SuNs9p/XbGcQezR7SguZrZzxqCQozxtd/N8UKBWbWjk=
+github.com/uudashr/gopkgs v2.0.1+incompatible/go.mod h1:MtCdKVJkxW7hNKWXPNWfpaeEp8+Ml3Q8myb4yWhn2Hg=
+github.com/uudashr/gopkgs/v2 v2.1.2 h1:A0+QH6wqNRHORJnxmqfeuBEsK4nYQ7pgcOHhqpqcrpo=
+github.com/uudashr/gopkgs/v2 v2.1.2/go.mod h1:O9VKOuPWrUpVhaxcg7N3QiTrlDhgJb/84Y7b3qaX1rI=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
 github.com/xdg/stringprep v1.0.0 h1:d9X0esnoa3dFsV0FG35rAT0RIhYFlPq7MiP+DW89La0=

chaincode/ledger_db.go

@@ -30,7 +30,7 @@ type State struct {
 // LedgerDB to access the chaincode database during the lifetime of a SmartContract
 type LedgerDB struct {
 	cc               shim.ChaincodeStubInterface
-	tuplesEvent      *TuplesEvent
+	event            *Event
 	transactionState State
 	mutex            *sync.RWMutex
 }
@@ -443,17 +443,17 @@ func (db *LedgerDB) GetNode(key string) (Node, error) {
 // High-level functions for events
 // ----------------------------------------------
 
-// SendTuplesEvent sends an event with updated tuples if there is any
+// SendEvent sends an event with updated tuples if there is any
 // Only one event can be sent per transaction
-func (db *LedgerDB) SendTuplesEvent() error {
-	if db.tuplesEvent == nil {
+func (db *LedgerDB) SendEvent() error {
+	if db.event == nil {
 		return nil
 	}
-	payload, err := json.Marshal(*(db.tuplesEvent))
+	payload, err := json.Marshal(*(db.event))
 	if err != nil {
 		return err
 	}
-	err = db.cc.SetEvent("tuples-updated", payload)
+	err = db.cc.SetEvent("chaincode-updates", payload)
 	if err != nil {
 		return err
 	}
@@ -472,8 +472,8 @@ func (db *LedgerDB) AddTupleEvent(tupleKey string) error {
 	if genericTuple.Status != StatusTodo {
 		return nil
 	}
-	if db.tuplesEvent == nil {
-		db.tuplesEvent = &TuplesEvent{}
+	if db.event == nil {
+		db.event = &Event{}
 	}
 	switch genericTuple.AssetType {
 	case TraintupleType:
@@ -483,31 +483,52 @@ func (db *LedgerDB) AddTupleEvent(tupleKey string) error {
 		}
 		out := outputTraintuple{}
 		out.Fill(db, tuple, tupleKey)
-		db.tuplesEvent.Traintuples = append(db.tuplesEvent.Traintuples, out)
+		db.event.Traintuples = append(db.event.Traintuples, out)
 	case CompositeTraintupleType:
 		tuple, err := db.GetCompositeTraintuple(tupleKey)
 		if err != nil {
 			return err
 		}
 		out := outputCompositeTraintuple{}
 		out.Fill(db, tuple, tupleKey)
-		db.tuplesEvent.CompositeTraintuples = append(db.tuplesEvent.CompositeTraintuples, out)
+		db.event.CompositeTraintuples = append(db.event.CompositeTraintuples, out)
 	case AggregatetupleType:
 		tuple, err := db.GetAggregatetuple(tupleKey)
 		if err != nil {
 			return err
 		}
 		out := outputAggregatetuple{}
 		out.Fill(db, tuple, tupleKey)
-		db.tuplesEvent.Aggregatetuples = append(db.tuplesEvent.Aggregatetuples, out)
+		db.event.Aggregatetuples = append(db.event.Aggregatetuples, out)
 	case TesttupleType:
 		tuple, err := db.GetTesttuple(tupleKey)
 		if err != nil {
 			return err
 		}
 		out := outputTesttuple{}
 		out.Fill(db, tupleKey, tuple)
-		db.tuplesEvent.Testtuples = append(db.tuplesEvent.Testtuples, out)
+		db.event.Testtuples = append(db.event.Testtuples, out)
 	}
 	return nil
 }
+
+// AddComputePlanEvent add the compute plan matching the ID to the event struct
+func (db *LedgerDB) AddComputePlanEvent(ComputePlanID, status string) error {
+	if !stringInSlice(status, []string{StatusCanceled, StatusFailed, StatusDone}) {
+		return nil
+	}
+	if db.event == nil {
+		db.event = &Event{}
+	}
+	cp := eventComputePlan{
+		ComputePlanID: ComputePlanID,
+		Status:        status,
+	}
+	algokeys, err := db.GetIndexKeys("algo~computeplanid~key", []string{"algo", ComputePlanID})
+	if err != nil {
+		return err
+	}
+	cp.AlgoKeys = algokeys
+	db.event.ComputePlans = append(db.event.ComputePlans, cp)
+	return nil
+}

chaincode/main.go

@@ -189,7 +189,7 @@ func (t *SubstraChaincode) Invoke(stub shim.ChaincodeStubInterface) peer.Respons
 	}
 	// Send event if there is any. It's done in one batch since we can only send
 	// one event per call
-	err = db.SendTuplesEvent()
+	err = db.SendEvent()
 	if err != nil {
 		return formatErrorResponse(errors.Internal("could not send event: %s", err.Error()))
 	}

chaincode/output.go

@@ -283,12 +283,19 @@ type outputModel struct {
 	Testtuple           outputTesttuple            `json:"testtuple"`
 }
 
-// TuplesEvent is the collection of tuples sent in an event
-type TuplesEvent struct {
+// Event is the collection of tuples sent in an event
+type Event struct {
 	Testtuples           []outputTesttuple           `json:"testtuple"`
 	Traintuples          []outputTraintuple          `json:"traintuple"`
 	CompositeTraintuples []outputCompositeTraintuple `json:"compositeTraintuple"`
 	Aggregatetuples      []outputAggregatetuple      `json:"aggregatetuple"`
+	ComputePlans         []eventComputePlan          `json:"computePlan"`
+}
+
+type eventComputePlan struct {
+	ComputePlanID string   `json:"computePlanID"`
+	Status        string   `json:"status"`
+	AlgoKeys      []string `json:"algoKeys"`
 }
 
 type outputComputePlan struct {

chaincode/traintuple.go

@@ -193,6 +193,9 @@ func (traintuple *Traintuple) Save(db *LedgerDB, traintupleKey string) error {
 		if err := db.CreateIndex("computePlan~computeplanid~worker~rank~key", []string{"computePlan", traintuple.ComputePlanID, traintuple.Dataset.Worker, strconv.Itoa(traintuple.Rank), traintupleKey}); err != nil {
 			return err
 		}
+		if err := db.CreateIndex("algo~computeplanid~key", []string{"algo", traintuple.ComputePlanID, traintuple.AlgoKey}); err != nil {
+			return err
+		}
 	}
 	if traintuple.Tag != "" {
 		err := db.CreateIndex("traintuple~tag~key", []string{"traintuple", traintuple.Tag, traintupleKey})

chaincode/traintuple_composite.go

@@ -240,6 +240,9 @@ func (traintuple *CompositeTraintuple) Save(db *LedgerDB, traintupleKey string)
 		if err := db.CreateIndex("computePlan~computeplanid~worker~rank~key", []string{"computePlan", traintuple.ComputePlanID, traintuple.Dataset.Worker, strconv.Itoa(traintuple.Rank), traintupleKey}); err != nil {
 			return err
 		}
+		if err := db.CreateIndex("algo~computeplanid~key", []string{"algo", traintuple.ComputePlanID, traintuple.AlgoKey}); err != nil {
+			return err
+		}
 	}
 	if traintuple.Tag != "" {
 		err := db.CreateIndex("compositeTraintuple~tag~key", []string{"compositeTraintuple", traintuple.Tag, traintupleKey})

chaincode/tuple_aggregate.go

@@ -203,6 +203,9 @@ func (tuple *Aggregatetuple) Save(db *LedgerDB, aggregatetupleKey string) error
 		if err := db.CreateIndex("computePlan~computeplanid~worker~rank~key", []string{"computePlan", tuple.ComputePlanID, tuple.Worker, strconv.Itoa(tuple.Rank), aggregatetupleKey}); err != nil {
 			return err
 		}
+		if err := db.CreateIndex("algo~computeplanid~key", []string{"algo", tuple.ComputePlanID, tuple.AlgoKey}); err != nil {
+			return err
+		}
 	}
 	if tuple.Tag != "" {
 		err := db.CreateIndex("aggregatetuple~tag~key", []string{"aggregatetuple", tuple.Tag, aggregatetupleKey})