Adjust timestamp in metrics post requests to account for endpoint's c…

…lock offset (CMC-1356) (#78)

check/checkresult_metrics.go

@@ -24,8 +24,9 @@ import (
 )
 
 // NewMetricsPostRequest function sets up a request with provided
-// check results set data
-func NewMetricsPostRequest(crs *ResultSet) *protocol.MetricsPostRequest {
+// check results set data. The timestamp within the request will be adjusted for the far-end's clock
+// offset given as clockOffset, in milliseconds.
+func NewMetricsPostRequest(crs *ResultSet, clockOffset int64) *protocol.MetricsPostRequest {
 	req := &protocol.MetricsPostRequest{}
 	req.Version = "1"
 	req.Method = "check_metrics.post_multi"
@@ -35,7 +36,7 @@ func NewMetricsPostRequest(crs *ResultSet) *protocol.MetricsPostRequest {
 	req.Params.MinCheckPeriod = crs.Check.GetPeriod() * 1000
 	req.Params.State = crs.State
 	req.Params.Status = crs.Status
-	req.Params.Timestamp = utils.NowTimestampMillis()
+	req.Params.Timestamp = utils.NowTimestampMillis() + clockOffset
 	if crs.Length() == 0 {
 		req.Params.Metrics = nil
 	} else {

check/checkresult_test.go

@@ -0,0 +1,58 @@
+//
+// Copyright 2017 Rackspace
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS-IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+package check_test
+
+import (
+	"context"
+	"github.com/racker/rackspace-monitoring-poller/check"
+	"github.com/racker/rackspace-monitoring-poller/protocol/metric"
+	"github.com/racker/rackspace-monitoring-poller/utils"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestNewMetricsPostRequest(t *testing.T) {
+	checkData := `{
+	  "id":"chTestTCP_TLSRunSuccess",
+	  "zone_id":"pzA",
+	  "entity_id":"enAAAAIPV4",
+	  "details":{"port":443,"ssl":true},
+	  "type":"remote.tcp",
+	  "timeout":15,
+	  "period":30,
+	  "ip_addresses":{"default":"127.0.0.1"},
+	  "target_alias":"default",
+	  "target_hostname":"",
+	  "target_resolver":"",
+	  "disabled":false
+	  }`
+	ch, err := check.NewCheck(context.Background(), []byte(checkData))
+	require.NoError(t, err)
+
+	cr := check.NewResult(metric.NewMetric("tt_connect", "", metric.MetricNumber, 500, metric.UnitMilliseconds))
+	crs := check.NewResultSet(ch, cr)
+
+	origTimeStamper := utils.InstallAlternateTimestampFunc(func() int64 {
+		return 5000
+	})
+	defer utils.InstallAlternateTimestampFunc(origTimeStamper)
+
+	req := check.NewMetricsPostRequest(crs, 200)
+
+	assert.Equal(t, int64(5200), req.Params.Timestamp)
+}

hostinfo/hostinfo_test.go

@@ -31,7 +31,9 @@ func TestHostInfoMemory_PopulateResult(t *testing.T) {
 	hostInfoMemory := hostinfo.NewHostInfoMemory(hinfo)
 	result, err := hostInfoMemory.Run()
 	sourceFrame := &protocol.FrameMsg{}
-	utils.NowTimestampMillis = func() int64 { return 100 }
+
+	origTs := utils.InstallAlternateTimestampFunc(func() int64 { return 100 })
+	defer utils.InstallAlternateTimestampFunc(origTs)
 	response := hostinfo.NewHostInfoResponse(result, sourceFrame)
 	encoded, err := response.Encode()
 	assert.NoError(t, err)

poller/connection_stream.go

@@ -139,7 +139,7 @@ func (cs *EleConnectionStream) SendMetrics(crs *check.ResultSet) error {
 	}
 
 	if conn := cs.conns.ChooseBest(); conn != nil {
-		conn.GetSession().Send(check.NewMetricsPostRequest(crs))
+		conn.GetSession().Send(check.NewMetricsPostRequest(crs, conn.GetClockOffset()))
 	}
 
 	return nil

poller/connection_stream_test.go

@@ -178,6 +178,7 @@ func TestEleConnectionStream_SendMetrics_Normal(t *testing.T) {
 
 	c2 := poller.NewMockConnection(ctrl)
 	c2.EXPECT().GetLatency().AnyTimes().Return(int64(10))
+	c2.EXPECT().GetClockOffset().AnyTimes().Return(int64(0))
 	c2.EXPECT().GetSession().Return(mockSession)
 
 	c3 := poller.NewMockConnection(ctrl)

poller/connection_stream_whitebox_test.go

@@ -117,6 +117,7 @@ func TestConnectionStream_SendMetrics(t *testing.T) {
 	defer mockCtrl.Finish()
 	mockSession := NewMockSession(mockCtrl)
 	mockSession.EXPECT().GetLatency().AnyTimes().Return(int64(0))
+	mockSession.EXPECT().GetClockOffset().AnyTimes().Return(int64(0))
 	ctx := context.Background()
 
 	basicCheck, err := check.NewCheck(ctx, []byte(`{

poller/session_test.go

@@ -86,15 +86,13 @@ func (fm frameMatcher) String() string {
 }
 
 func installDeterministicTimestamper(startingTimestamp, timestampInc int64) utils.NowTimestampMillisFunc {
-	origTimestamper := utils.NowTimestampMillis
 	mockTimestamp := startingTimestamp
-	utils.NowTimestampMillis = func() int64 {
+
+	return utils.InstallAlternateTimestampFunc(func() int64 {
 		ts := mockTimestamp
 		mockTimestamp += timestampInc
 		return ts
-	}
-
-	return origTimestamper
+	})
 }
 
 func prepareHandshakeResponse(heartbeatInterval uint64, readsHere io.Writer) {
@@ -126,7 +124,7 @@ func TestEleSession_HeartbeatSending(t *testing.T) {
 	defer readsHere.Close()
 
 	origTimestamper := installDeterministicTimestamper(1000, 2000)
-	defer func() { utils.NowTimestampMillis = origTimestamper }()
+	defer utils.InstallAlternateTimestampFunc(origTimestamper)
 
 	const heartbeatInterval = 10 // ms
 
@@ -168,7 +166,7 @@ func TestEleSession_HeartbeatConsumption(t *testing.T) {
 	defer readsHere.Close()
 
 	origTimestamper := installDeterministicTimestamper(1000, 2000)
-	defer func() { utils.NowTimestampMillis = origTimestamper }()
+	defer utils.InstallAlternateTimestampFunc(origTimestamper)
 
 	const heartbeatInterval = 10 // ms
 
@@ -409,7 +407,7 @@ func TestEleSession_PollerPrepare(t *testing.T) {
 			defer readsHere.Close()
 
 			origTimestamper := installDeterministicTimestamper(1000, 2000)
-			defer func() { utils.NowTimestampMillis = origTimestamper }()
+			defer utils.InstallAlternateTimestampFunc(origTimestamper)
 
 			es := poller.NewSession(context.Background(), eleConn, reconciler, &config.Config{})
 			defer es.Close()

utils/timestamp.go

@@ -27,6 +27,14 @@ var NowTimestampMillis NowTimestampMillisFunc = func() int64 {
 	return time.Now().UnixNano() / int64(time.Millisecond)
 }
 
+// InstallAlternateTimestampFunc is intended for unit testing where a deterministic timestamp needs to be
+// temporarily enabled. Be sure to defer re-invoke this function to re-install the prior one.
+func InstallAlternateTimestampFunc(newFunc NowTimestampMillisFunc) (priorFunc NowTimestampMillisFunc) {
+	priorFunc = NowTimestampMillis
+	NowTimestampMillis = newFunc
+	return
+}
+
 // ScaleFractionalDuration is primarily useful when scaling durations that are "sub second", but more generally
 // it's when duration is smaller than targetUnits. In that case, a fractional value is much more meangingful than
 // a 0, which is what would happen with plain duration (i.e. integer) division. targetUnits should really be