/
metricsender.go
104 lines (94 loc) · 2.94 KB
/
metricsender.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
// Copyright 2014 Canonical Ltd.
// Licensed under the AGPLv3, see LICENCE file for details.
// Package metricsender contains functions for sending
// metrics from a state server to a remote metric collector.
package metricsender
import (
"time"
"github.com/juju/errors"
"github.com/juju/loggo"
"github.com/juju/juju/apiserver/metricsender/wireformat"
"github.com/juju/juju/state"
)
var logger = loggo.GetLogger("juju.apiserver.metricsender")
// MetricSender defines the interface used to send metrics
// to a collection service.
type MetricSender interface {
Send([]*wireformat.MetricBatch) (*wireformat.Response, error)
}
func handleResponse(mm *state.MetricsManager, st *state.State, response wireformat.Response) {
for _, envResp := range response.EnvResponses {
err := st.SetMetricBatchesSent(envResp.AcknowledgedBatches)
if err != nil {
logger.Errorf("failed to set sent on metrics %v", err)
}
for unitName, status := range envResp.UnitStatuses {
unit, err := st.Unit(unitName)
if err != nil {
logger.Errorf("failed to retrieve unit %q: %v", unitName, err)
continue
}
err = unit.SetMeterStatus(status.Status, status.Info)
if err != nil {
logger.Errorf("failed to set unit %q meter status to %v: %v", unitName, status, err)
}
}
}
if response.NewGracePeriod > 0 {
err := mm.SetGracePeriod(response.NewGracePeriod)
if err != nil {
logger.Errorf("failed to set new grace period %v", err)
}
}
}
// SendMetrics will send any unsent metrics
// over the MetricSender interface in batches
// no larger than batchSize.
func SendMetrics(st *state.State, sender MetricSender, batchSize int) error {
metricsManager, err := st.MetricsManager()
if err != nil {
return errors.Trace(err)
}
for {
metrics, err := st.MetricsToSend(batchSize)
if err != nil {
return errors.Trace(err)
}
if len(metrics) == 0 {
logger.Infof("nothing to send")
break
}
wireData := make([]*wireformat.MetricBatch, len(metrics))
for i, m := range metrics {
wireData[i] = wireformat.ToWire(m)
}
response, err := sender.Send(wireData)
if err != nil {
logger.Errorf("%+v", err)
if incErr := metricsManager.IncrementConsecutiveErrors(); incErr != nil {
logger.Errorf("failed to increment error count %v", incErr)
return errors.Trace(errors.Wrap(err, incErr))
}
return errors.Trace(err)
}
if response != nil {
// TODO (mattyw) We are currently ignoring errors during response handling.
handleResponse(metricsManager, st, *response)
if err := metricsManager.SetLastSuccessfulSend(time.Now()); err != nil {
err = errors.Annotate(err, "failed to set successful send time")
logger.Warningf("%v", err)
return errors.Trace(err)
}
}
}
unsent, err := st.CountOfUnsentMetrics()
if err != nil {
return errors.Trace(err)
}
sent, err := st.CountOfSentMetrics()
if err != nil {
return errors.Trace(err)
}
logger.Infof("metrics collection summary: sent:%d unsent:%d", sent, unsent)
return nil
}