Skip to content

Commit

Permalink
vtorc: Switch to Vitess stats (#15948)
Browse files Browse the repository at this point in the history
Signed-off-by: Dirkjan Bussink <d.bussink@gmail.com>
  • Loading branch information
dbussink committed May 17, 2024
1 parent 3b1800d commit 9911880
Show file tree
Hide file tree
Showing 7 changed files with 28 additions and 54 deletions.
1 change: 0 additions & 1 deletion go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,6 @@ require (
github.com/planetscale/vtprotobuf v0.5.0
github.com/prometheus/client_golang v1.19.0
github.com/prometheus/common v0.53.0
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
github.com/sjmudd/stopwatch v0.1.1
github.com/soheilhy/cmux v0.1.5
github.com/spf13/cobra v1.8.0
Expand Down
2 changes: 0 additions & 2 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -420,8 +420,6 @@ github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsT
github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A=
github.com/prometheus/procfs v0.14.0 h1:Lw4VdGGoKEZilJsayHf0B+9YgLGREba2C6xr+Fdfq6s=
github.com/prometheus/procfs v0.14.0/go.mod h1:XL+Iwz8k8ZabyZfMFHPiilCniixqQarAy5Mu67pHlNQ=
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM=
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
github.com/richardartoul/molecule v1.0.1-0.20221107223329-32cfee06a052 h1:Qp27Idfgi6ACvFQat5+VJvlYToylpM/hcyLBI3WaKPA=
Expand Down
18 changes: 7 additions & 11 deletions go/vt/vtorc/inst/analysis_dao.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,30 +21,26 @@ import (
"math"
"time"

"vitess.io/vitess/go/vt/external/golib/sqlutils"
"vitess.io/vitess/go/vt/log"
"vitess.io/vitess/go/vt/topo/topoproto"

"github.com/patrickmn/go-cache"
"google.golang.org/protobuf/encoding/prototext"

"vitess.io/vitess/go/stats"
"vitess.io/vitess/go/vt/external/golib/sqlutils"
"vitess.io/vitess/go/vt/log"
topodatapb "vitess.io/vitess/go/vt/proto/topodata"
"vitess.io/vitess/go/vt/topo"
"vitess.io/vitess/go/vt/topo/topoproto"
"vitess.io/vitess/go/vt/vtctl/reparentutil"
"vitess.io/vitess/go/vt/vtorc/config"
"vitess.io/vitess/go/vt/vtorc/db"
"vitess.io/vitess/go/vt/vtorc/util"

"github.com/patrickmn/go-cache"
"github.com/rcrowley/go-metrics"
)

var analysisChangeWriteCounter = metrics.NewCounter()
var analysisChangeWriteCounter = stats.NewCounter("analysis.change.write", "Number of times analysis has changed")

var recentInstantAnalysis *cache.Cache

func init() {
_ = metrics.Register("analysis.change.write", analysisChangeWriteCounter)

go initializeAnalysisDaoPostConfiguration()
}

Expand Down Expand Up @@ -717,7 +713,7 @@ func auditInstanceAnalysisInChangelog(tabletAlias string, analysisCode AnalysisC
tabletAlias, string(analysisCode),
)
if err == nil {
analysisChangeWriteCounter.Inc(1)
analysisChangeWriteCounter.Add(1)
} else {
log.Error(err)
}
Expand Down
7 changes: 3 additions & 4 deletions go/vt/vtorc/inst/analysis_dao_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import (
"time"

"github.com/patrickmn/go-cache"
"github.com/rcrowley/go-metrics"
"github.com/stretchr/testify/require"

"vitess.io/vitess/go/vt/external/golib/sqlutils"
Expand Down Expand Up @@ -906,7 +905,7 @@ func TestAuditInstanceAnalysisInChangelog(t *testing.T) {
oldAnalysisChangeWriteCounter := analysisChangeWriteCounter

recentInstantAnalysis = cache.New(tt.cacheExpiration, 100*time.Millisecond)
analysisChangeWriteCounter = metrics.NewCounter()
before := analysisChangeWriteCounter.Get()

defer func() {
// Set the old values back.
Expand All @@ -919,7 +918,7 @@ func TestAuditInstanceAnalysisInChangelog(t *testing.T) {
updates := []struct {
tabletAlias string
analysisCode AnalysisCode
writeCounterExpectation int
writeCounterExpectation int64
wantErr string
}{
{
Expand Down Expand Up @@ -950,7 +949,7 @@ func TestAuditInstanceAnalysisInChangelog(t *testing.T) {
continue
}
require.NoError(t, err)
require.EqualValues(t, upd.writeCounterExpectation, analysisChangeWriteCounter.Count())
require.EqualValues(t, upd.writeCounterExpectation, analysisChangeWriteCounter.Get()-before)
}
})
}
Expand Down
11 changes: 3 additions & 8 deletions go/vt/vtorc/inst/audit_dao.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,18 +21,13 @@ import (
"os"
"time"

"github.com/rcrowley/go-metrics"

"vitess.io/vitess/go/stats"
"vitess.io/vitess/go/vt/log"
"vitess.io/vitess/go/vt/vtorc/config"
"vitess.io/vitess/go/vt/vtorc/db"
)

var auditOperationCounter = metrics.NewCounter()

func init() {
_ = metrics.Register("audit.write", auditOperationCounter)
}
var auditOperationCounter = stats.NewCounter("audit.write", "Number of audit operations performed")

// AuditOperation creates and writes a new audit entry by given params
func AuditOperation(auditType string, tabletAlias string, message string) error {
Expand Down Expand Up @@ -86,7 +81,7 @@ func AuditOperation(auditType string, tabletAlias string, message string) error
if !auditWrittenToFile {
log.Infof(logMessage)
}
auditOperationCounter.Inc(1)
auditOperationCounter.Add(1)

return nil
}
Expand Down
15 changes: 4 additions & 11 deletions go/vt/vtorc/inst/instance_dao.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@ import (
"time"

"github.com/patrickmn/go-cache"
"github.com/rcrowley/go-metrics"
"github.com/sjmudd/stopwatch"

"vitess.io/vitess/go/mysql/replication"
Expand Down Expand Up @@ -60,10 +59,8 @@ var (
var forgetAliases *cache.Cache

var (
accessDeniedCounter = metrics.NewCounter()
readTopologyInstanceCounter = metrics.NewCounter()
readInstanceCounter = metrics.NewCounter()
writeInstanceCounter = metrics.NewCounter()
readTopologyInstanceCounter = stats.NewCounter("instance.read_topology", "Number of times an instance was read from the topology")
readInstanceCounter = stats.NewCounter("instance.read", "Number of times an instance was read")
backendWrites = collection.CreateOrReturnCollection("BACKEND_WRITES")
writeBufferLatency = stopwatch.NewNamedStopwatch()
)
Expand All @@ -74,10 +71,6 @@ var (
)

func init() {
_ = metrics.Register("instance.access_denied", accessDeniedCounter)
_ = metrics.Register("instance.read_topology", readTopologyInstanceCounter)
_ = metrics.Register("instance.read", readInstanceCounter)
_ = metrics.Register("instance.write", writeInstanceCounter)
_ = writeBufferLatency.AddMany([]string{"wait", "write"})
writeBufferLatency.Start("wait")

Expand Down Expand Up @@ -390,7 +383,7 @@ Cleanup:
}

latency.Stop("instance")
readTopologyInstanceCounter.Inc(1)
readTopologyInstanceCounter.Add(1)

if instanceFound {
instance.LastDiscoveryLatency = time.Since(readingStartTime)
Expand Down Expand Up @@ -621,7 +614,7 @@ func ReadInstance(tabletAlias string) (*Instance, bool, error) {
instances, err := readInstancesByCondition(condition, sqlutils.Args(tabletAlias), "")
// We know there will be at most one (alias is the PK).
// And we expect to find one.
readInstanceCounter.Inc(1)
readInstanceCounter.Add(1)
if len(instances) == 0 {
return nil, false, err
}
Expand Down
28 changes: 11 additions & 17 deletions go/vt/vtorc/logic/vtorc.go
Original file line number Diff line number Diff line change
Expand Up @@ -25,9 +25,9 @@ import (
"time"

"github.com/patrickmn/go-cache"
"github.com/rcrowley/go-metrics"
"github.com/sjmudd/stopwatch"

"vitess.io/vitess/go/stats"
"vitess.io/vitess/go/vt/log"
"vitess.io/vitess/go/vt/servenv"
"vitess.io/vitess/go/vt/vtorc/collection"
Expand All @@ -50,32 +50,26 @@ var snapshotDiscoveryKeys chan string
var snapshotDiscoveryKeysMutex sync.Mutex
var hasReceivedSIGTERM int32

var discoveriesCounter = metrics.NewCounter()
var failedDiscoveriesCounter = metrics.NewCounter()
var instancePollSecondsExceededCounter = metrics.NewCounter()
var discoveryQueueLengthGauge = metrics.NewGauge()
var discoveryRecentCountGauge = metrics.NewGauge()
var discoveriesCounter = stats.NewCounter("discoveries.attempt", "Number of discoveries attempted")
var failedDiscoveriesCounter = stats.NewCounter("discoveries.fail", "Number of failed discoveries")
var instancePollSecondsExceededCounter = stats.NewCounter("discoveries.instance_poll_seconds_exceeded", "Number of instances that took longer than InstancePollSeconds to poll")
var discoveryQueueLengthGauge = stats.NewGauge("discoveries.queue_length", "Length of the discovery queue")
var discoveryRecentCountGauge = stats.NewGauge("discoveries.recent_count", "Number of recent discoveries")
var discoveryMetrics = collection.CreateOrReturnCollection(DiscoveryMetricsName)

var recentDiscoveryOperationKeys *cache.Cache

func init() {
snapshotDiscoveryKeys = make(chan string, 10)

_ = metrics.Register("discoveries.attempt", discoveriesCounter)
_ = metrics.Register("discoveries.fail", failedDiscoveriesCounter)
_ = metrics.Register("discoveries.instance_poll_seconds_exceeded", instancePollSecondsExceededCounter)
_ = metrics.Register("discoveries.queue_length", discoveryQueueLengthGauge)
_ = metrics.Register("discoveries.recent_count", discoveryRecentCountGauge)

ometrics.OnMetricsTick(func() {
discoveryQueueLengthGauge.Update(int64(discoveryQueue.QueueLen()))
discoveryQueueLengthGauge.Set(int64(discoveryQueue.QueueLen()))
})
ometrics.OnMetricsTick(func() {
if recentDiscoveryOperationKeys == nil {
return
}
discoveryRecentCountGauge.Update(int64(recentDiscoveryOperationKeys.ItemCount()))
discoveryRecentCountGauge.Set(int64(recentDiscoveryOperationKeys.ItemCount()))
})
}

Expand Down Expand Up @@ -168,7 +162,7 @@ func DiscoverInstance(tabletAlias string, forceDiscovery bool) {
latency.Stop("total")
discoveryTime := latency.Elapsed("total")
if discoveryTime > instancePollSecondsDuration() {
instancePollSecondsExceededCounter.Inc(1)
instancePollSecondsExceededCounter.Add(1)
log.Warningf("discoverInstance exceeded InstancePollSeconds for %+v, took %.4fs", tabletAlias, discoveryTime.Seconds())
if metric != nil {
metric.InstancePollSecondsDurationCount = 1
Expand Down Expand Up @@ -196,7 +190,7 @@ func DiscoverInstance(tabletAlias string, forceDiscovery bool) {
return
}

discoveriesCounter.Inc(1)
discoveriesCounter.Add(1)

// First we've ever heard of this instance. Continue investigation:
instance, err := inst.ReadTopologyInstanceBufferable(tabletAlias, latency)
Expand All @@ -211,7 +205,7 @@ func DiscoverInstance(tabletAlias string, forceDiscovery bool) {
}

if instance == nil {
failedDiscoveriesCounter.Inc(1)
failedDiscoveriesCounter.Add(1)
metric = &discovery.Metric{
Timestamp: time.Now(),
TabletAlias: tabletAlias,
Expand Down

0 comments on commit 9911880

Please sign in to comment.