VTOrc: Update the primary key for all the tables from hostname, port to alias

go/vt/vtorc/inst/analysis_dao.go

@@ -38,13 +38,11 @@ import (
 	"github.com/rcrowley/go-metrics"
 )
 
-var analysisChangeWriteAttemptCounter = metrics.NewCounter()
 var analysisChangeWriteCounter = metrics.NewCounter()
 
 var recentInstantAnalysis *cache.Cache
 
 func init() {
-	_ = metrics.Register("analysis.change.write.attempt", analysisChangeWriteAttemptCounter)
 	_ = metrics.Register("analysis.change.write", analysisChangeWriteCounter)
 
 	go initializeAnalysisDaoPostConfiguration()
@@ -611,11 +609,8 @@ func auditInstanceAnalysisInChangelog(tabletAlias string, analysisCode AnalysisC
 			return nil
 		}
 	}
-	// Passed the cache; but does database agree that there's a change? Here's a persistent cache; this comes here
-	// to verify no two vtorc services are doing this without coordinating (namely, one dies, the other taking its place
-	// and has no familiarity of the former's cache)
-	analysisChangeWriteAttemptCounter.Inc(1)
 
+	// Find if the lastAnalysisHasChanged or not while updating the row if it has.
 	lastAnalysisChanged := false
 	{
 		sqlResult, err := db.ExecVTOrc(`
@@ -637,10 +632,15 @@ func auditInstanceAnalysisInChangelog(tabletAlias string, analysisCode AnalysisC
 			log.Error(err)
 			return err
 		}
-		lastAnalysisChanged = (rows > 0)
+		lastAnalysisChanged = rows > 0
 	}
+
+	// If the last analysis has not changed, then there is a chance that this is the first insertion.
+	// We need to find that out too when we insert into the database.
+	firstInsertion := false
 	if !lastAnalysisChanged {
-		_, err := db.ExecVTOrc(`
+		// The insert only returns more than 1 row changed if this is the first insertion.
+		sqlResult, err := db.ExecVTOrc(`
 			insert ignore into database_instance_last_analysis (
 					alias, analysis_timestamp, analysis
 				) values (
@@ -653,9 +653,16 @@ func auditInstanceAnalysisInChangelog(tabletAlias string, analysisCode AnalysisC
 			log.Error(err)
 			return err
 		}
+		rows, err := sqlResult.RowsAffected()
+		if err != nil {
+			log.Error(err)
+			return err
+		}
+		firstInsertion = rows > 0
 	}
 	recentInstantAnalysis.Set(tabletAlias, analysisCode, cache.DefaultExpiration)
-	if !lastAnalysisChanged {
+	// If the analysis has changed or if it is the first insertion, we need to make sure we write this change to the database.
+	if !lastAnalysisChanged && !firstInsertion {
 		return nil
 	}
 

go/vt/vtorc/inst/analysis_dao_test.go

@@ -18,7 +18,10 @@ package inst
 
 import (
 	"testing"
+	"time"
 
+	"github.com/patrickmn/go-cache"
+	"github.com/rcrowley/go-metrics"
 	"github.com/stretchr/testify/require"
 
 	"vitess.io/vitess/go/vt/external/golib/sqlutils"
@@ -672,3 +675,75 @@ func TestGetReplicationAnalysis(t *testing.T) {
 		})
 	}
 }
+
+// TestAuditInstanceAnalysisInChangelog tests the functionality of the auditInstanceAnalysisInChangelog function
+// and verifies that we write the correct amount of times to the database.
+func TestAuditInstanceAnalysisInChangelog(t *testing.T) {
+	tests := []struct {
+		name            string
+		cacheExpiration time.Duration
+	}{
+		{
+			name:            "Long expiration",
+			cacheExpiration: 2 * time.Minute,
+		}, {
+			name:            "Very short expiration",
+			cacheExpiration: 100 * time.Millisecond,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create the cache for the test to use.
+			oldRecentInstantAnalysisCache := recentInstantAnalysis
+			oldAnalysisChangeWriteCounter := analysisChangeWriteCounter
+
+			recentInstantAnalysis = cache.New(tt.cacheExpiration, 100*time.Millisecond)
+			analysisChangeWriteCounter = metrics.NewCounter()
+
+			defer func() {
+				// Set the old values back.
+				recentInstantAnalysis = oldRecentInstantAnalysisCache
+				analysisChangeWriteCounter = oldAnalysisChangeWriteCounter
+				// Each test should clear the database. The easiest way to do that is to run all the initialization commands again.
+				db.ClearVTOrcDatabase()
+			}()
+
+			updates := []struct {
+				tabletAlias             string
+				analysisCode            AnalysisCode
+				writeCounterExpectation int
+				wantErr                 string
+			}{
+				{
+					// Store a new analysis for the zone1-100 tablet.
+					tabletAlias:             "zone1-100",
+					analysisCode:            ReplicationStopped,
+					writeCounterExpectation: 1,
+				}, {
+					// Write the same analysis, no new write should happen.
+					tabletAlias:             "zone1-100",
+					analysisCode:            ReplicationStopped,
+					writeCounterExpectation: 1,
+				}, {
+					// Change the analysis. This should trigger an update.
+					tabletAlias:             "zone1-100",
+					analysisCode:            ReplicaSemiSyncMustBeSet,
+					writeCounterExpectation: 2,
+				},
+			}
+
+			for _, upd := range updates {
+				// We sleep 200 milliseconds to make sure that the cache has had time to update.
+				// It should be able to delete entries if the expiration is less than 200 milliseconds.
+				time.Sleep(200 * time.Millisecond)
+				err := auditInstanceAnalysisInChangelog(upd.tabletAlias, upd.analysisCode)
+				if upd.wantErr != "" {
+					require.EqualError(t, err, upd.wantErr)
+					continue
+				}
+				require.NoError(t, err)
+				require.EqualValues(t, upd.writeCounterExpectation, analysisChangeWriteCounter.Count())
+			}
+		})
+	}
+}