From 40158eda71ef9716e7f3bed6a61e05999e0174c2 Mon Sep 17 00:00:00 2001 From: James Read Date: Wed, 14 Aug 2024 23:06:50 +0100 Subject: [PATCH] bugfix: Race condition on executor logs (#385) --- internal/executor/executor.go | 68 +++++++++++++++++++++++++++++++---- internal/grpcapi/grpcApi.go | 21 ++++++----- 2 files changed, 74 insertions(+), 15 deletions(-) diff --git a/internal/executor/executor.go b/internal/executor/executor.go index a3151a5..95d8514 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -37,9 +37,11 @@ type ActionBinding struct { // Executor represents a helper class for executing commands. It's main method // is ExecRequest type Executor struct { - Logs map[string]*InternalLogEntry + logs map[string]*InternalLogEntry LogsByActionId map[string][]*InternalLogEntry + logmutex sync.RWMutex + MapActionIdToBinding map[string]*ActionBinding MapActionIdToBindingLock sync.RWMutex @@ -100,7 +102,7 @@ type executorStepFunc func(*ExecutionRequest) bool func DefaultExecutor(cfg *config.Config) *Executor { e := Executor{} e.Cfg = cfg - e.Logs = make(map[string]*InternalLogEntry) + e.logs = make(map[string]*InternalLogEntry) e.LogsByActionId = make(map[string][]*InternalLogEntry) e.MapActionIdToBinding = make(map[string]*ActionBinding) @@ -132,6 +134,52 @@ func (e *Executor) AddListener(m listener) { e.listeners = append(e.listeners, m) } +func (e *Executor) GetLogsCopy() map[string]*InternalLogEntry { + e.logmutex.RLock() + + copy := make(map[string]*InternalLogEntry) + + for k, v := range e.logs { + copy[k] = v + } + + e.logmutex.RUnlock() + + return copy +} + +func (e *Executor) GetLog(trackingID string) (*InternalLogEntry, bool) { + e.logmutex.RLock() + + entry, found := e.logs[trackingID] + + e.logmutex.RUnlock() + + return entry, found +} + +func (e *Executor) GetLogsByActionId(actionId string) []*InternalLogEntry { + e.logmutex.RLock() + + logs, found := e.LogsByActionId[actionId] + + e.logmutex.RUnlock() + + if !found { + return make([]*InternalLogEntry, 0) + } + + return logs +} + +func (e *Executor) SetLog(trackingID string, entry *InternalLogEntry) { + e.logmutex.Lock() + + e.logs[trackingID] = entry + + e.logmutex.Unlock() +} + // ExecRequest processes an ExecutionRequest func (e *Executor) ExecRequest(req *ExecutionRequest) (*sync.WaitGroup, string) { req.executor = e @@ -147,7 +195,7 @@ func (e *Executor) ExecRequest(req *ExecutionRequest) (*sync.WaitGroup, string) ActionIcon: "💩", } - _, isDuplicate := e.Logs[req.TrackingID] + _, isDuplicate := e.GetLog(req.TrackingID) if isDuplicate || req.TrackingID == "" { req.TrackingID = uuid.NewString() @@ -155,7 +203,7 @@ func (e *Executor) ExecRequest(req *ExecutionRequest) (*sync.WaitGroup, string) log.Tracef("executor.ExecRequest(): %v", req) - e.Logs[req.TrackingID] = req.logEntry + e.SetLog(req.TrackingID, req.logEntry) wg := new(sync.WaitGroup) wg.Add(1) @@ -185,12 +233,16 @@ func (e *Executor) execChain(req *ExecutionRequest) { func getConcurrentCount(req *ExecutionRequest) int { concurrentCount := 0 - for _, log := range req.executor.LogsByActionId[req.Action.ID] { + req.executor.logmutex.RLock() + + for _, log := range req.executor.GetLogsByActionId(req.Action.ID) { if !log.ExecutionFinished { concurrentCount += 1 } } + req.executor.logmutex.RUnlock() + return concurrentCount } @@ -232,7 +284,7 @@ func getExecutionsCount(rate config.RateSpec, req *ExecutionRequest) int { then := time.Now().Add(-duration) - for _, logEntry := range req.executor.LogsByActionId[req.Action.ID] { + for _, logEntry := range req.executor.GetLogsByActionId(req.Action.ID) { if logEntry.DatetimeStarted.After(then) && !logEntry.Blocked { executions += 1 @@ -308,12 +360,16 @@ func stepRequestAction(req *ExecutionRequest) bool { req.logEntry.ActionIcon = req.Action.Icon req.logEntry.ActionId = req.Action.ID + req.executor.logmutex.Lock() + if _, containsKey := req.executor.LogsByActionId[req.Action.ID]; !containsKey { req.executor.LogsByActionId[req.Action.ID] = make([]*InternalLogEntry, 0) } req.executor.LogsByActionId[req.Action.ID] = append(req.executor.LogsByActionId[req.Action.ID], req.logEntry) + req.executor.logmutex.Unlock() + log.WithFields(log.Fields{ "actionTitle": req.logEntry.ActionTitle, "tags": req.Tags, diff --git a/internal/grpcapi/grpcApi.go b/internal/grpcapi/grpcApi.go index 91806b3..22c2b18 100644 --- a/internal/grpcapi/grpcApi.go +++ b/internal/grpcapi/grpcApi.go @@ -35,7 +35,7 @@ func (api *oliveTinAPI) KillAction(ctx ctx.Context, req *pb.KillActionRequest) ( ExecutionTrackingId: req.ExecutionTrackingId, } - execReqLogEntry, found := api.executor.Logs[req.ExecutionTrackingId] + execReqLogEntry, found := api.executor.GetLog(req.ExecutionTrackingId) ret.Found = found @@ -99,7 +99,7 @@ func (api *oliveTinAPI) StartActionAndWait(ctx ctx.Context, req *pb.StartActionA wg, _ := api.executor.ExecRequest(&execReq) wg.Wait() - internalLogEntry, ok := api.executor.Logs[execReq.TrackingID] + internalLogEntry, ok := api.executor.GetLog(execReq.TrackingID) if ok { return &pb.StartActionAndWaitResponse{ @@ -142,7 +142,7 @@ func (api *oliveTinAPI) StartActionByGetAndWait(ctx ctx.Context, req *pb.StartAc wg, _ := api.executor.ExecRequest(&execReq) wg.Wait() - internalLogEntry, ok := api.executor.Logs[execReq.TrackingID] + internalLogEntry, ok := api.executor.GetLog(execReq.TrackingID) if ok { return &pb.StartActionByGetAndWaitResponse{ @@ -172,7 +172,7 @@ func internalLogEntryToPb(logEntry *executor.InternalLogEntry) *pb.LogEntry { } func getExecutionStatusByTrackingID(api *oliveTinAPI, executionTrackingId string) *executor.InternalLogEntry { - logEntry, ok := api.executor.Logs[executionTrackingId] + logEntry, ok := api.executor.GetLog(executionTrackingId) if !ok { return nil @@ -184,10 +184,13 @@ func getExecutionStatusByTrackingID(api *oliveTinAPI, executionTrackingId string func getMostRecentExecutionStatusById(api *oliveTinAPI, actionId string) *executor.InternalLogEntry { var ile *executor.InternalLogEntry - for _, candidateLe := range api.executor.Logs { - if actionId == candidateLe.ActionId { - ile = candidateLe - } + logs := api.executor.GetLogsByActionId(actionId) + + if len(logs) == 0 { + return nil + } else { + // Get last log entry + ile = logs[len(logs)-1] } return ile @@ -262,7 +265,7 @@ func (api *oliveTinAPI) GetLogs(ctx ctx.Context, req *pb.GetLogsRequest) (*pb.Ge // TODO Limit to 10 entries or something to prevent browser lag. - for trackingId, logEntry := range api.executor.Logs { + for trackingId, logEntry := range api.executor.GetLogsCopy() { action := cfg.FindAction(logEntry.ActionTitle) if action == nil || acl.IsAllowedLogs(cfg, user, action) {