-
Notifications
You must be signed in to change notification settings - Fork 24
/
adapter.go
315 lines (266 loc) · 9.69 KB
/
adapter.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
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
/*
Copyright 2020 VMware, Inc.
SPDX-License-Identifier: Apache-2.0
*/
package vsphere
import (
"context"
"fmt"
"time"
cloudevents "github.com/cloudevents/sdk-go/v2"
"github.com/jpillora/backoff"
"github.com/vmware/govmomi"
"github.com/vmware/govmomi/event"
"github.com/vmware/govmomi/vim25/methods"
"github.com/vmware/govmomi/vim25/types"
"go.uber.org/zap"
"knative.dev/eventing/pkg/adapter/v2"
"knative.dev/pkg/kvstore"
"knative.dev/pkg/logging"
kubeclient "knative.dev/pkg/client/injection/kube/client"
)
const (
// signal unstable event API for converting vSphere events to CE
eventTypeFormat = "com.vmware.vsphere.%s.v0"
// extended attribute to filter on vSphere API version/class
ceVSphereAPIKey = "vsphereapiversion"
ceVSphereEventClass = "eventclass"
// read up to max events per iteration
maxEventsBatch = 100
)
type envConfig struct {
adapter.EnvConfig
// KVConfigMap is the name of the configmap to use as our kvstore.
KVConfigMap string `envconfig:"VSPHERE_KVSTORE_CONFIGMAP" required:"true"`
// CheckpointConfig configures the checkpoint behavior of this controller
CheckpointConfig string `envconfig:"VSPHERE_CHECKPOINT_CONFIG" default:"{}"`
// PayloadEncoding configures the encoding format for the cloud event payload
PayloadEncoding string `envconfig:"VSPHERE_PAYLOAD_ENCODING" default:"application/xml"`
}
func NewEnvConfig() adapter.EnvConfigAccessor {
return &envConfig{}
}
// vAdapter implements the vSphereSource adapter to trigger a Sink.
type vAdapter struct {
Logger *zap.SugaredLogger
Namespace string
Source string
VClient *govmomi.Client
VAPIVersion string
CEClient cloudevents.Client
KVStore kvstore.Interface
CpConfig CheckpointConfig
PayloadEncoding string
}
func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClient cloudevents.Client) adapter.Adapter {
env := processed.(*envConfig)
logger := logging.FromContext(ctx)
vClient, err := NewSOAPClient(ctx)
if err != nil {
logger.Fatalf("unable to create vSphere client: %v", err)
}
source := vClient.URL().String()
if source == "" {
logger.Fatal("unable to determine vSphere client source: empty host")
}
// setup checkpointing
store := kvstore.NewConfigMapKVStore(ctx, env.KVConfigMap, env.Namespace, kubeclient.Get(ctx).CoreV1())
if err = store.Init(ctx); err != nil {
logger.Fatalf("could not initialize kv store: %v", err)
}
cpconf, err := newCheckpointConfig(env.CheckpointConfig)
if err != nil {
logger.Fatalf("could not not read checkpoint config: %v", err)
}
logger.Infow("configuring checkpointing", zap.String("ReplayWindow", cpconf.MaxAge.String()),
zap.String("Period", cpconf.Period.String()))
if cpconf.MaxAge == time.Duration(0) {
logger.Warn("disabling event replay: maxAge set to 0s")
}
return &vAdapter{
Logger: logger,
Namespace: env.Namespace,
Source: source,
VClient: vClient,
VAPIVersion: vClient.ServiceContent.About.ApiVersion,
CEClient: ceClient,
KVStore: store,
CpConfig: *cpconf,
PayloadEncoding: env.PayloadEncoding,
}
}
// Start implements adapter.Adapter
func (a *vAdapter) Start(ctx context.Context) error {
defer func() {
// using fresh ctx to avoid canceled error during logout
_ = a.VClient.Logout(context.Background()) // best effort, ignoring error
}()
return a.run(ctx)
}
// run will start reading events from vCenter and send them to the configured
// sink. The internal vCenter event (history) collector will attempt to replay
// events starting at the current vCenter time or retrieved from a previous
// checkpoint with additional validation logic to avoid unbounded event replay.
// A checkpoint will be created periodically to track the position in the
// vCenter event stream. This allows to implement at-least-once semantics.
func (a *vAdapter) run(ctx context.Context) error {
var cp checkpoint
if err := a.KVStore.Get(ctx, checkpointKey, &cp); err != nil {
logging.FromContext(ctx).Warnw("could not retrieve checkpoint configuration", zap.Error(err))
}
// begin of event stream defaults to current vCenter time (UTC)
vcTime, err := methods.GetCurrentTime(ctx, a.VClient)
if err != nil {
return fmt.Errorf("get current time from vCenter: %w", err)
}
begin := getBeginFromCheckpoint(ctx, *vcTime, cp, a.CpConfig.MaxAge)
coll, err := newHistoryCollector(ctx, a.VClient.Client, begin)
if err != nil {
return fmt.Errorf("create event collector: %w", err)
}
return a.readEvents(ctx, coll)
}
// readEvents polls vCenter for new events starting at the configured begin time
// in the provided event history collector. A checkpoint will be periodically
// created and stored in Kubernetes to track successfully processed events
// (ACK-ed by sink).
func (a *vAdapter) readEvents(ctx context.Context, c *event.HistoryCollector) error {
logger := logging.FromContext(ctx)
var (
lastEvent types.BaseEvent
lastCheckpointEventKey int32
)
bOff := backoff.Backoff{
Factor: 2,
Jitter: false,
Min: time.Second,
Max: 5 * time.Second,
}
cpTicker := time.NewTicker(a.CpConfig.Period)
defer cpTicker.Stop()
for {
select {
case <-ctx.Done():
return ctx.Err()
// checkpoints
case <-cpTicker.C:
// avoid unnecessary K8s API calls
skip := lastEvent == nil || lastCheckpointEventKey == lastEvent.GetEvent().Key
if !skip {
var current checkpoint
if err := a.KVStore.Get(ctx, checkpointKey, ¤t); err != nil {
return fmt.Errorf("retrieve current checkpoint: %w", err)
}
logger.Debugw("creating checkpoint", zap.Any("checkpoint", current))
if err := a.KVStore.Save(ctx); err != nil {
return fmt.Errorf("save checkpoint: %w", err)
}
lastCheckpointEventKey = lastEvent.GetEvent().Key
} else {
logger.Debug("skipping checkpoint: no new events since last checkpoint")
}
// poll vCenter events
default:
events, err := c.ReadNextEvents(ctx, maxEventsBatch)
if err != nil {
return fmt.Errorf("read events from vcenter: %w", err)
}
if len(events) == 0 {
delay := bOff.Duration()
logger.Debugw("backing off retrieving events: no new events received", zap.Duration("backoffSeconds", delay))
time.Sleep(delay)
continue
}
logger.Debugf("got %d events", len(events))
n, err := a.sendEvents(ctx, events)
if err != nil {
// TODO: return and fail instead?
logger.Errorf("send events: success %d (total %d): %v", n, len(events), err)
// special case: all events failed so skipping checkpoint
if n == 0 {
continue
}
}
if n == 0 && err == nil {
panic("we should never get here")
}
// last successfully sent event from batch
lastEvent = events[n-1]
cp := checkpoint{
VCenter: a.Source,
LastEventKey: lastEvent.GetEvent().Key,
LastEventType: getEventDetails(lastEvent).Type,
LastEventKeyTimestamp: lastEvent.GetEvent().CreatedTime,
CreatedTimestamp: time.Now().UTC(),
}
if err = a.KVStore.Set(ctx, checkpointKey, cp); err != nil {
return fmt.Errorf("set checkpoint: %w", err)
}
bOff.Reset()
}
}
}
// sendEvents converts all events to cloud events and sends them to the
// configured sink. It returns the number of successfully processed events,
// which might 0, partial or all events. sendEvents returns when all events are
// processed or on the first error.
func (a *vAdapter) sendEvents(ctx context.Context, baseEvents []types.BaseEvent) (int, error) {
var success int
for _, be := range baseEvents {
ev := cloudevents.NewEvent(cloudevents.VersionV1)
ev.SetSource(a.Source)
details := getEventDetails(be)
// CE envelop
ev.SetID(fmt.Sprintf("%d", be.GetEvent().Key))
ev.SetType(fmt.Sprintf(eventTypeFormat, details.Type))
ev.SetTime(be.GetEvent().CreatedTime)
ev.SetExtension(ceVSphereEventClass, details.Class)
ev.SetExtension(ceVSphereAPIKey, a.VAPIVersion)
if err := ev.SetData(a.PayloadEncoding, be); err != nil {
return success, fmt.Errorf("set data on event: %w", err)
}
// TODO: better partial batch failure handling here?
logging.FromContext(ctx).Debugw("sending event",
zap.String("ID", ev.ID()),
zap.String("type", ev.Type()),
zap.Any("data", be),
)
result := a.CEClient.Send(ctx, ev)
if !cloudevents.IsACK(result) {
logging.FromContext(ctx).Errorw("failed to send cloudevent", zap.Error(result))
return success, result
}
success++
}
return success, nil
}
// getBeginFromCheckpoint returns the valid begin time to start replaying
// vCenter events. If the checkpoint is empty the current vCenter time (UTC) is
// used. If the last checkpoint event timestamp is larger than maxAge, replay
// will start at maxAge.
func getBeginFromCheckpoint(ctx context.Context, vcTime time.Time, cp checkpoint, maxAge time.Duration) time.Time {
begin := vcTime
logger := logging.FromContext(ctx)
cpTime := cp.LastEventKeyTimestamp
if !cpTime.IsZero() {
// valid checkpoint
logger.Info("found existing checkpoint")
maxTime := begin.Add(maxAge * -1)
if maxTime.Unix() > cpTime.Unix() {
logger.Warnw("potential data loss: last event timestamp in checkpoint is older than configured maximum",
zap.String("maxHistory", maxAge.String()), zap.String("checkpointTimestamp",
cp.LastEventKeyTimestamp.String()))
begin = maxTime
logger.Warnw("setting begin of event stream", zap.String("beginTimestamp", begin.String()))
} else {
begin = cpTime
logger.Infow("setting begin of event stream", zap.String("beginTimestamp", begin.String()),
zap.Int32("eventKey", cp.LastEventKey))
}
} else {
// empty checkpoint
logger.Info("no valid checkpoint found")
logger.Infow("setting begin of event stream", zap.String("beginTimestamp", begin.String()))
}
return begin
}