tweaks/hacks to allow performance measurements

go.sum

@@ -126,6 +126,7 @@ github.com/hashicorp/serf v0.10.1 h1:Z1H2J60yRKvfDYAOZLd2MU0ND4AH/WDz7xYHDWQsIPY
 github.com/hashicorp/serf v0.10.1/go.mod h1:yL2t6BqATOLGc5HF7qbFkTfXoPIY0WZdWHfEvMqbG+4=
 github.com/hazelcast/hazelcast-go-client v1.4.1 h1:BSpJqqjbACI4MugfWXGxk+JdZR3JRELx0n769pa85kA=
 github.com/hazelcast/hazelcast-go-client v1.4.1/go.mod h1:PJ38lqXJ18S0YpkrRznPDlUH8GnnMAQCx3jpQtBPZ6Q=
+github.com/hazelcast/hazelcast-go-client v1.4.2/go.mod h1:PJ38lqXJ18S0YpkrRznPDlUH8GnnMAQCx3jpQtBPZ6Q=
 github.com/invopop/yaml v0.2.0 h1:7zky/qH+O0DwAyoobXUqvVBwgBFRxKoQ/3FjcVpjTMY=
 github.com/invopop/yaml v0.2.0/go.mod h1:2XuRLgs/ouIrW3XNzuNj7J3Nvu/Dig5MXvbCEdiBN3Q=
 github.com/ioki-mobility/go-outline v0.5.0 h1:BsMcH/wbfcgGpH1+8VTsygzdh9rQmUL3aMpcYW0MKuw=

openapi/api.yaml

@@ -293,6 +293,9 @@ components:
         createdAt:
           type: string
           format: date-time
+        completedAt:
+          type: string
+          format: date-time
         state:
           type: string
           enum:

pkg/bpmn_engine/api/server/server.go

@@ -184,13 +184,15 @@ func (s *server) GetProcessInstance(ctx echo.Context, processInstanceKey int64)
 
 	key := fmt.Sprintf("%d", pi.Key)
 	processDefintionKey := fmt.Sprintf("%d", pi.ProcessDefinitionKey)
-	time := time.Unix(0, pi.CreatedAt*int64(time.Second))
+	createdAt := time.Unix(0, pi.CreatedAt*int64(time.Second))
+	completedAt := time.Unix(0, pi.CompletedAt*int64(time.Second))
 	state := api.ProcessInstanceState(fmt.Sprintf("%d", pi.State))
 	processInstanceSimple := api.ProcessInstance{
 		Key:                  &key,
 		ProcessDefinitionKey: &processDefintionKey,
 		State:                &state,
-		CreatedAt:            &time,
+		CreatedAt:            &createdAt,
+		CompletedAt:          &completedAt,
 		CaughtEvents:         &pi.CaughtEvents,
 		VariableHolder:       &pi.VariableHolder,
 		Activities:           &pi.Activities,
@@ -216,13 +218,15 @@ func (s *server) GetProcessInstances(ctx echo.Context, params api.GetProcessInst
 		for _, pi := range partitionProcessInstances {
 			key := fmt.Sprintf("%d", pi.Key)
 			processDefintionKey := fmt.Sprintf("%d", pi.ProcessDefinitionKey)
-			time := time.Unix(0, pi.CreatedAt*int64(time.Second))
+			createdAt := time.Unix(0, pi.CreatedAt*int64(time.Second))
+			completedAt := time.Unix(0, pi.CompletedAt*int64(time.Second))
 			state := api.ProcessInstanceState(fmt.Sprintf("%d", pi.State))
 			processInstanceSimple := api.ProcessInstance{
 				Key:                  &key,
 				ProcessDefinitionKey: &processDefintionKey,
 				State:                &state,
-				CreatedAt:            &time,
+				CreatedAt:            &createdAt,
+				CompletedAt:          &completedAt,
 				CaughtEvents:         &pi.CaughtEvents,
 				VariableHolder:       &pi.VariableHolder,
 				Activities:           &pi.Activities,

pkg/bpmn_engine/engine.go

@@ -240,6 +240,10 @@ func (state *BpmnEngineState) run(instance *processInstanceInfo) (err error) {
 		// TODO need to send failed State
 		state.exportEndProcessEvent(*process, *instance)
 	}
+	if instance.State == Completed {
+		instance.CompletedAt = time.Now()
+		instance.SetVariable("duration_in_ms", fmt.Sprintf("%d", time.Since(instance.CreatedAt).Milliseconds()))
+	}
 	// TODO: persistently update state
 	state.persistence.PersistProcessInstance(instance)
 

pkg/bpmn_engine/jobs.go

@@ -70,9 +70,17 @@ func (state *BpmnEngineState) handleServiceTask(process *ProcessInfo, instance *
 	state.persistence.PersistJob(job)
 	// }
 
+	// FIXME: remove this one
+	autoComplete(state, job)
+
 	return job.JobState == Completed, job
 }
 
+func autoComplete(state *BpmnEngineState, j *job) {
+	j.JobState = Completed
+	state.persistence.PersistJob(j)
+}
+
 func (state *BpmnEngineState) JobCompleteById(jobId int64) {
 	jobs := state.persistence.FindJobs("", nil, jobId)
 

pkg/bpmn_engine/persistence/rqlite/persistence_rqlite.go

@@ -23,7 +23,7 @@ type BpmnEnginePersistenceRqlite struct {
 func NewBpmnEnginePersistenceRqlite(cfg *Config) *BpmnEnginePersistenceRqlite {
 	context := Start(cfg)
 
-	time.Sleep(2 * time.Second)
+	time.Sleep(2 * time.Second) // FIXME: don't depend on time, rather using a loop, too check if port 4001 is already open
 
 	Init(context.Str)
 
@@ -136,14 +136,15 @@ func (persistence *BpmnEnginePersistenceRqlite) FindProcessInstances(processInst
 			def := new(sql.ProcessInstanceEntity)
 
 			def.Key = (*r.Parameters[0]).GetI()
-			def.ProcessDefinitionKey = int64((*r.Parameters[1]).GetI())
+			def.ProcessDefinitionKey = (*r.Parameters[1]).GetI()
 			def.CreatedAt = (*r.Parameters[2]).GetI()
+			def.CompletedAt = (*r.Parameters[3]).GetI()
 
-			def.State = int((*r.Parameters[3]).GetI())
-			def.VariableHolder = (*r.Parameters[4]).GetS()
-			def.CaughtEvents = (*r.Parameters[5]).GetS()
+			def.State = int((*r.Parameters[4]).GetI())
+			def.VariableHolder = (*r.Parameters[5]).GetS()
+			def.CaughtEvents = (*r.Parameters[6]).GetS()
 
-			def.Activities = (*r.Parameters[6]).GetS()
+			def.Activities = (*r.Parameters[7]).GetS()
 
 			processInstances = append(processInstances, def)
 

pkg/bpmn_engine/persistence/rqlite/sql/process_instance.go

@@ -6,6 +6,7 @@ type ProcessInstanceEntity struct {
 	Key                  int64
 	ProcessDefinitionKey int64
 	CreatedAt            int64
+	CompletedAt          int64
 	State                int
 	VariableHolder       string
 	CaughtEvents         string
@@ -17,6 +18,7 @@ CREATE TABLE IF NOT EXISTS process_instance (
 	key INTEGER PRIMARY KEY,
 	process_definition_key INTEGER NOT NULL,
 	created_at INTEGER NOT NULL,
+	completed_at INTEGER,
 	state INTEGER NOT NULL,
 	variable_holder TEXT NOT NULL,
 	caught_events TEXT NOT NULL,
@@ -26,13 +28,25 @@ CREATE TABLE IF NOT EXISTS process_instance (
 
 const PROCESS_INSTANCE_INSERT = `
 	INSERT INTO process_instance
-	(key, process_definition_key, created_at, state, variable_holder, caught_events, activities)
+	(key, process_definition_key, created_at, completed_at, state, variable_holder, caught_events, activities)
 	VALUES
-	(%d, %d, %d, %d, '%s', '%s', '%s') ON CONFLICT DO UPDATE SET state = %d, variable_holder = '%s', caught_events = '%s', activities = '%s';`
+	(%d, %d, %d, %d, %d, '%s', '%s', '%s') ON CONFLICT DO UPDATE SET state = %d, variable_holder = '%s', caught_events = '%s', activities = '%s';`
 
-const PROCESS_INSTANCE_SELECT = `SELECT key, process_definition_key, created_at, state, variable_holder, caught_events, activities FROM process_instance WHERE %s ORDER BY created_at DESC;`
+const PROCESS_INSTANCE_SELECT = `SELECT key, process_definition_key, created_at, completed_at, state, variable_holder, caught_events, activities FROM process_instance WHERE %s ORDER BY created_at DESC;`
 
 func BuildProcessInstanceUpsertQuery(pie *ProcessInstanceEntity) string {
 	// FIXME: for speed this needs to be splited
-	return fmt.Sprintf(PROCESS_INSTANCE_INSERT, pie.Key, pie.ProcessDefinitionKey, pie.CreatedAt, pie.State, pie.VariableHolder, pie.CaughtEvents, pie.Activities, pie.State, pie.VariableHolder, pie.CaughtEvents, pie.Activities)
+	return fmt.Sprintf(PROCESS_INSTANCE_INSERT,
+		pie.Key,
+		pie.ProcessDefinitionKey,
+		pie.CreatedAt,
+		pie.CompletedAt,
+		pie.State,
+		pie.VariableHolder,
+		pie.CaughtEvents,
+		pie.Activities,
+		pie.State,
+		pie.VariableHolder,
+		pie.CaughtEvents,
+		pie.Activities)
 }

pkg/bpmn_engine/process_instance.go

@@ -12,6 +12,7 @@ type processInstanceInfo struct {
 	InstanceKey    int64                     `json:"ik"`
 	VariableHolder var_holder.VariableHolder `json:"vh,omitempty"`
 	CreatedAt      time.Time                 `json:"c"`
+	CompletedAt    time.Time                 `json:"cl"`
 	State          ActivityState             `json:"s"`
 	CaughtEvents   []catchEvent              `json:"ce,omitempty"`
 	activities     []activity

pkg/bpmn_engine/rqlite_persistence_facade.go

@@ -321,6 +321,7 @@ func (persistence *BpmnEnginePersistenceRqlite) PersistProcessInstance(processIn
 		Key:                  processInstance.InstanceKey,
 		ProcessDefinitionKey: processInstance.ProcessInfo.ProcessKey,
 		CreatedAt:            processInstance.CreatedAt.Unix(),
+		CompletedAt:          processInstance.CompletedAt.Unix(),
 		State:                activityStateMap[processInstance.State],
 		VariableHolder:       string(varaiblesJson),
 		CaughtEvents:         string(caughtEvents),