feat: Separate Queued Tables from In Progress Tables (#920)

bbernays · web-flow · commit dcb5d26b3ee2 · 2023-05-31T12:48:56.000Z
#### Summary

When users limit the concurrency of their sync, tables that have not been started will appear to be "in-progress" even though they are waiting to start...

This PR separates `queued` tables from `in-progress` by defining `in-progress` tables as those that have no `end-time`, but do have a non-zero start time. `queued` tables are those tables that have a zero start and end time
diff --git a/plugins/source/metrics.go b/plugins/source/metrics.go
@@ -172,8 +172,9 @@ func (s *Metrics) InProgressTables() []string {
 		for _, clientMetrics := range tableMetrics {
 			clientMetrics.mutex.Lock()
 			endTime := clientMetrics.endTime
+			startTime := clientMetrics.startTime
 			clientMetrics.mutex.Unlock()
-			if endTime.IsZero() {
+			if endTime.IsZero() && !startTime.IsZero() {
 				inProgressTables = append(inProgressTables, table)
 				break
 			}
@@ -184,3 +185,23 @@ func (s *Metrics) InProgressTables() []string {
 
 	return inProgressTables
 }
+
+func (s *Metrics) QueuedTables() []string {
+	var queuedTables []string
+
+	for table, tableMetrics := range s.TableClient {
+		for _, clientMetrics := range tableMetrics {
+			clientMetrics.mutex.Lock()
+			startTime := clientMetrics.startTime
+			endTime := clientMetrics.endTime
+			clientMetrics.mutex.Unlock()
+			if startTime.IsZero() && endTime.IsZero() {
+				queuedTables = append(queuedTables, table)
+				break
+			}
+		}
+	}
+
+	slices.Sort(queuedTables)
+	return queuedTables
+}
diff --git a/plugins/source/metrics_test.go b/plugins/source/metrics_test.go
@@ -71,10 +71,44 @@ func TestInProgressTables(t *testing.T) {
 		Panics:    3,
 		startTime: time.Now(),
 	}
-
+	s.TableClient["test_table_running3"] = make(map[string]*TableClientMetrics)
+	s.TableClient["test_table_running3"]["testExecutionClient"] = &TableClientMetrics{}
 	assert.ElementsMatch(t, []string{"test_table_running1", "test_table_running2"}, s.InProgressTables())
 }
 
+func TestQueuedTables(t *testing.T) {
+	s := &Metrics{
+		TableClient: make(map[string]map[string]*TableClientMetrics),
+	}
+	s.TableClient["test_table_done"] = make(map[string]*TableClientMetrics)
+	s.TableClient["test_table_done"]["testExecutionClient"] = &TableClientMetrics{
+		Resources: 1,
+		Errors:    2,
+		Panics:    3,
+		startTime: time.Now(),
+		endTime:   time.Now().Add(time.Second),
+	}
+
+	s.TableClient["test_table_running1"] = make(map[string]*TableClientMetrics)
+	s.TableClient["test_table_running1"]["testExecutionClient"] = &TableClientMetrics{
+		Resources: 1,
+		Errors:    2,
+		Panics:    3,
+		startTime: time.Now(),
+	}
+
+	s.TableClient["test_table_running2"] = make(map[string]*TableClientMetrics)
+	s.TableClient["test_table_running2"]["testExecutionClient"] = &TableClientMetrics{
+		Resources: 1,
+		Errors:    2,
+		Panics:    3,
+		startTime: time.Now(),
+	}
+	s.TableClient["test_table_running3"] = make(map[string]*TableClientMetrics)
+	s.TableClient["test_table_running3"]["testExecutionClient"] = &TableClientMetrics{}
+	assert.ElementsMatch(t, []string{"test_table_running3"}, s.QueuedTables())
+}
+
 type MockClientMeta struct {
 }
 
diff --git a/plugins/source/scheduler.go b/plugins/source/scheduler.go
@@ -146,17 +146,22 @@ func (p *Plugin) periodicMetricLogger(ctx context.Context, wg *sync.WaitGroup) {
 			return
 		case <-ticker.C:
 			inProgressTables := p.metrics.InProgressTables()
-
+			queuedTables := p.metrics.QueuedTables()
 			logLine := p.logger.Info().
 				Uint64("total_resources", p.metrics.TotalResourcesAtomic()).
 				Uint64("total_errors", p.metrics.TotalErrorsAtomic()).
 				Uint64("total_panics", p.metrics.TotalPanicsAtomic()).
-				Int("num_in_progress_tables", len(inProgressTables))
+				Int("num_in_progress_tables", len(inProgressTables)).
+				Int("num_queued_tables", len(queuedTables))
 
 			if len(inProgressTables) <= periodicMetricLoggerLogTablesLimit {
 				logLine.Strs("in_progress_tables", inProgressTables)
 			}
 
+			if len(queuedTables) <= periodicMetricLoggerLogTablesLimit {
+				logLine.Strs("queued_tables", queuedTables)
+			}
+
 			logLine.Msg("Sync in progress")
 		}
 	}
