Update stats report interval to send each second

cmd/di.go

@@ -526,6 +526,7 @@ func (di *Dependencies) bootstrapNodeComponents(nodeOptions node.Options, tequil
 		connectivity.NewStatusSender(),
 		di.IPResolver,
 		connection.DefaultIPCheckParams(),
+		connection.DefaultStatsReportInterval,
 	)
 
 	di.LogCollector = logconfig.NewCollector(&logconfig.CurrentLogOptions)

core/connection/manager.go

@@ -104,6 +104,7 @@ type connectionManager struct {
 	connectivityStatusSender connectivity.StatusSender
 	ipResolver               ip.Resolver
 	ipCheckParams            IPCheckParams
+	statsReportInterval      time.Duration
 
 	// These are populated by Connect at runtime.
 	ctx                    context.Context
@@ -127,6 +128,7 @@ func NewManager(
 	connectivityStatusSender connectivity.StatusSender,
 	ipResolver ip.Resolver,
 	ipCheckParams IPCheckParams,
+	statsReportInterval time.Duration,
 ) *connectionManager {
 	return &connectionManager{
 		newDialog:                dialogCreator,
@@ -138,6 +140,7 @@ func NewManager(
 		cleanup:                  make([]func() error, 0),
 		ipResolver:               ipResolver,
 		ipCheckParams:            ipCheckParams,
+		statsReportInterval:      statsReportInterval,
 	}
 }
 
@@ -389,7 +392,7 @@ func (manager *connectionManager) startConnection(
 		return err
 	}
 
-	statsPublisher := newStatsPublisher(manager.eventPublisher)
+	statsPublisher := newStatsPublisher(manager.eventPublisher, manager.statsReportInterval)
 	go statsPublisher.start(manager.getCurrentSession(), conn)
 
 	manager.cleanup = append(manager.cleanup, func() error {

core/connection/manager_test.go

@@ -44,6 +44,7 @@ type testContext struct {
 	fakeResolver          ip.Resolver
 	ipCheckParams         IPCheckParams
 	statusSender          *mockStatusSender
+	statsReportInterval   time.Duration
 	sync.RWMutex
 }
 
@@ -110,6 +111,7 @@ func (tc *testContext) SetupTest() {
 
 	tc.statusSender = &mockStatusSender{}
 	tc.fakeResolver = ip.NewResolverMock("ip")
+	tc.statsReportInterval = 1 * time.Millisecond
 
 	tc.connManager = NewManager(
 		dialogCreator,
@@ -128,6 +130,7 @@ func (tc *testContext) SetupTest() {
 		tc.statusSender,
 		tc.fakeResolver,
 		tc.ipCheckParams,
+		tc.statsReportInterval,
 	)
 }
 
@@ -336,11 +339,23 @@ func (tc *testContext) Test_ManagerPublishesEvents() {
 	assert.NoError(tc.T(), err)
 
 	waitABit()
-	time.Sleep(StatsReportInterval)
 
 	history := tc.stubPublisher.GetEventHistory()
-	assert.Len(tc.T(), history, 4)
+	assert.True(tc.T(), len(history) >= 4)
+
+	// Check if published to all expected topics.
+	expectedTopics := [...]string{AppTopicConsumerStatistics, AppTopicConsumerConnectionState, AppTopicConsumerSession}
+	for _, v := range expectedTopics {
+		var published bool
+		for _, h := range history {
+			if v == h.calledWithTopic {
+				published = true
+			}
+		}
+		tc.Assert().Truef(published, "expected publish event to %s", v)
+	}
 
+	// Check received events data.
 	for _, v := range history {
 		if v.calledWithTopic == AppTopicConsumerStatistics {
 			event := v.calledWithData.(SessionStatsEvent)

core/connection/stats_publisher.go

@@ -24,8 +24,8 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-// StatsReportInterval is interval for consumer connection statistics reporting.
-const StatsReportInterval = 2 * time.Second
+// DefaultStatsReportInterval is interval for consumer connection statistics reporting.
+const DefaultStatsReportInterval = 1 * time.Second
 
 type statsSupplier interface {
 	Statistics() (Statistics, error)
@@ -37,11 +37,11 @@ type statsPublisher struct {
 	interval time.Duration
 }
 
-func newStatsPublisher(bus eventbus.Publisher) statsPublisher {
+func newStatsPublisher(bus eventbus.Publisher, interval time.Duration) statsPublisher {
 	return statsPublisher{
 		done:     make(chan struct{}),
 		bus:      bus,
-		interval: StatsReportInterval,
+		interval: interval,
 	}
 }
 

services/openvpn/client.go

@@ -74,7 +74,7 @@ func NewClient(openvpnBinary, configDirectory, runtimeDirectory string,
 
 		stateMiddleware := newStateMiddleware(stateCh)
 		authMiddleware := newAuthMiddleware(options.SessionID, signer)
-		byteCountMiddleware := openvpn_bytescount.NewMiddleware(client.OnStats, connection.StatsReportInterval)
+		byteCountMiddleware := openvpn_bytescount.NewMiddleware(client.OnStats, connection.DefaultStatsReportInterval)
 		proc := openvpn.CreateNewProcess(openvpnBinary, vpnClientConfig.GenericConfig, stateMiddleware, byteCountMiddleware, authMiddleware)
 		return proc, vpnClientConfig, nil
 	}