forked from cloudfoundry/bosh-agent
-
Notifications
You must be signed in to change notification settings - Fork 0
/
nats_handler.go
212 lines (169 loc) · 4.86 KB
/
nats_handler.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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
package mbus
import (
"encoding/json"
"errors"
"fmt"
"net"
"net/url"
"os"
"os/signal"
"strings"
"sync"
"syscall"
"github.com/cloudfoundry/yagnats"
boshhandler "github.com/cloudfoundry/bosh-agent/handler"
boshplatform "github.com/cloudfoundry/bosh-agent/platform"
boshsettings "github.com/cloudfoundry/bosh-agent/settings"
bosherr "github.com/cloudfoundry/bosh-utils/errors"
boshlog "github.com/cloudfoundry/bosh-utils/logger"
)
const (
responseMaxLength = 1024 * 1024
)
type Handler interface {
Run(boshhandler.Func) error
Start(boshhandler.Func) error
RegisterAdditionalFunc(boshhandler.Func)
Send(target boshhandler.Target, topic boshhandler.Topic, message interface{}) error
Stop()
}
type natsHandler struct {
settingsService boshsettings.Service
client yagnats.NATSClient
platform boshplatform.Platform
handlerFuncs []boshhandler.Func
handlerFuncsLock sync.Mutex
logger boshlog.Logger
logTag string
}
func NewNatsHandler(
settingsService boshsettings.Service,
client yagnats.NATSClient,
logger boshlog.Logger,
platform boshplatform.Platform,
) Handler {
return &natsHandler{
settingsService: settingsService,
client: client,
platform: platform,
logger: logger,
logTag: "NATS Handler",
}
}
func (h *natsHandler) Run(handlerFunc boshhandler.Func) error {
err := h.Start(handlerFunc)
defer h.Stop()
if err != nil {
return bosherr.WrapError(err, "Starting nats handler")
}
h.runUntilInterrupted()
return nil
}
func (h *natsHandler) Start(handlerFunc boshhandler.Func) error {
h.RegisterAdditionalFunc(handlerFunc)
connProvider, err := h.getConnectionInfo()
if err != nil {
return bosherr.WrapError(err, "Getting connection info")
}
h.client.BeforeConnectCallback(func() {
hostSplit := strings.Split(connProvider.Addr, ":")
ip := hostSplit[0]
if net.ParseIP(ip) == nil {
return
}
err = h.platform.DeleteARPEntryWithIP(ip)
if err != nil {
h.logger.Error(h.logTag, "Cleaning ip-mac address cache for: %s", ip)
}
})
err = h.client.Connect(connProvider)
if err != nil {
return bosherr.WrapError(err, "Connecting")
}
settings := h.settingsService.GetSettings()
subject := fmt.Sprintf("agent.%s", settings.AgentID)
h.logger.Info(h.logTag, "Subscribing to %s", subject)
_, err = h.client.Subscribe(subject, func(natsMsg *yagnats.Message) {
// Do not lock handler funcs around possible network calls!
h.handlerFuncsLock.Lock()
handlerFuncs := h.handlerFuncs
h.handlerFuncsLock.Unlock()
for _, handlerFunc := range handlerFuncs {
h.handleNatsMsg(natsMsg, handlerFunc)
}
})
if err != nil {
return bosherr.WrapErrorf(err, "Subscribing to %s", subject)
}
return nil
}
func (h *natsHandler) RegisterAdditionalFunc(handlerFunc boshhandler.Func) {
// Currently not locking since RegisterAdditionalFunc
// is not a primary way of adding handlerFunc.
h.handlerFuncsLock.Lock()
h.handlerFuncs = append(h.handlerFuncs, handlerFunc)
h.handlerFuncsLock.Unlock()
}
func (h *natsHandler) Send(target boshhandler.Target, topic boshhandler.Topic, message interface{}) error {
bytes, err := json.Marshal(message)
if err != nil {
return bosherr.WrapErrorf(err, "Marshalling message (target=%s, topic=%s): %#v", target, topic, message)
}
h.logger.Info(h.logTag, "Sending %s message '%s'", target, topic)
h.logger.DebugWithDetails(h.logTag, "Message Payload", string(bytes))
settings := h.settingsService.GetSettings()
subject := fmt.Sprintf("%s.agent.%s.%s", target, topic, settings.AgentID)
return h.client.Publish(subject, bytes)
}
func (h *natsHandler) Stop() {
h.client.Disconnect()
}
func (h *natsHandler) handleNatsMsg(natsMsg *yagnats.Message, handlerFunc boshhandler.Func) {
respBytes, req, err := boshhandler.PerformHandlerWithJSON(
natsMsg.Payload,
handlerFunc,
responseMaxLength,
h.logger,
)
if err != nil {
h.logger.Error(h.logTag, "Running handler: %s", err)
return
}
if len(respBytes) > 0 {
err = h.client.Publish(req.ReplyTo, respBytes)
if err != nil {
h.logger.Error(h.logTag, "Publishing to the client: %s", err.Error())
}
}
}
func (h *natsHandler) runUntilInterrupted() {
defer h.client.Disconnect()
keepRunning := true
c := make(chan os.Signal, 1)
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
for keepRunning {
select {
case <-c:
keepRunning = false
}
}
}
func (h *natsHandler) getConnectionInfo() (*yagnats.ConnectionInfo, error) {
settings := h.settingsService.GetSettings()
natsURL, err := url.Parse(settings.Mbus)
if err != nil {
return nil, bosherr.WrapError(err, "Parsing Nats URL")
}
connInfo := new(yagnats.ConnectionInfo)
connInfo.Addr = natsURL.Host
user := natsURL.User
if user != nil {
password, passwordIsSet := user.Password()
if !passwordIsSet {
return nil, errors.New("No password set for connection")
}
connInfo.Password = password
connInfo.Username = user.Username()
}
return connInfo, nil
}