-
Notifications
You must be signed in to change notification settings - Fork 5
/
client.go
139 lines (129 loc) · 3.22 KB
/
client.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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
package client
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
"net/netip"
"strings"
"time"
)
const (
HomeUrl = "https://deployments.magistrala.abstractmachines.fr/telemetry"
stopWaitTime = 5 * time.Second
callHomeSleepTime = 30 * time.Minute
backOff = 10 * time.Second
apiKey = "77e04a7c-f207-40dd-8950-c344871fd516"
)
var ipEndpoints = []string{
"https://checkip.amazonaws.com/",
"https://ipinfo.io/ip",
"https://api.ipify.org/",
}
type homingService struct {
serviceName string
version string
logger *slog.Logger
cancel context.CancelFunc
httpClient http.Client
}
func New(svc, version string, homingLogger *slog.Logger, cancel context.CancelFunc) *homingService {
return &homingService{
serviceName: svc,
version: version,
logger: homingLogger,
cancel: cancel,
httpClient: *http.DefaultClient,
}
}
func (hs *homingService) CallHome(ctx context.Context) {
for {
select {
case <-ctx.Done():
hs.Stop()
default:
data := telemetryData{
Service: hs.serviceName,
Version: hs.version,
LastSeen: time.Now(),
}
for _, endpoint := range ipEndpoints {
ip, err := hs.getIP(endpoint)
if err != nil {
hs.logger.Warn(fmt.Sprintf("failed to obtain service public IP address for sending Magistrala usage telemetry with error: %v", err))
continue
}
ip = strings.ReplaceAll(ip, "\n", "")
ip = strings.ReplaceAll(ip, "\\", "")
parsedIP, err := netip.ParseAddr(ip)
if err != nil {
hs.logger.Warn(fmt.Sprintf("failed to parse ip address with error: %v", err))
continue
}
data.IpAddress = parsedIP.String()
break
}
if err := hs.send(&data); err != nil && data.IpAddress != "" {
hs.logger.Warn(fmt.Sprintf("failed to send Magistrala telemetry data with error: %v", err))
time.Sleep(backOff)
continue
}
}
time.Sleep(callHomeSleepTime)
}
}
func (hs *homingService) Stop() {
defer hs.cancel()
c := make(chan bool)
defer close(c)
select {
case <-c:
case <-time.After(stopWaitTime):
}
hs.logger.Info("call home service shutdown")
}
type telemetryData struct {
Service string `json:"service"`
IpAddress string `json:"ip_address"`
Version string `json:"magistrala_version"`
LastSeen time.Time `json:"last_seen"`
}
func (hs *homingService) getIP(endpoint string) (string, error) {
req, err := http.NewRequest(http.MethodGet, endpoint, nil)
if err != nil {
return "", err
}
res, err := hs.httpClient.Do(req)
if err != nil {
return "", err
}
b, err := io.ReadAll(res.Body)
defer res.Body.Close()
if err != nil {
return "", err
}
return string(b), nil
}
func (hs *homingService) send(telDat *telemetryData) error {
b, err := json.Marshal(telDat)
if err != nil {
return err
}
req, err := http.NewRequest(http.MethodPost, HomeUrl, bytes.NewReader(b))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("apikey", apiKey)
res, err := hs.httpClient.Do(req)
if err != nil || res.StatusCode != http.StatusCreated {
if res != nil {
return fmt.Errorf("unsuccessful sending telemetry data with code %d and error: %v", res.StatusCode, err)
}
return err
}
return nil
}