-
Notifications
You must be signed in to change notification settings - Fork 1.1k
/
host.go
434 lines (379 loc) 路 15.9 KB
/
host.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
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
// Copyright 2016-2018, Pulumi Corporation.
//
// 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 plugin
import (
"os"
"github.com/blang/semver"
"github.com/hashicorp/go-multierror"
"github.com/pkg/errors"
"github.com/pulumi/pulumi/pkg/diag"
"github.com/pulumi/pulumi/pkg/resource"
"github.com/pulumi/pulumi/pkg/tokens"
"github.com/pulumi/pulumi/pkg/util/cmdutil"
"github.com/pulumi/pulumi/pkg/util/contract"
"github.com/pulumi/pulumi/pkg/util/logging"
"github.com/pulumi/pulumi/pkg/workspace"
)
// A Host hosts provider plugins and makes them easily accessible by package name.
type Host interface {
// ServerAddr returns the address at which the host's RPC interface may be found.
ServerAddr() string
// Log logs a message, including errors and warnings. Messages can have a resource URN
// associated with them. If no urn is provided, the message is global.
Log(sev diag.Severity, urn resource.URN, msg string, streamID int32)
// LogStatus logs a status message message, including errors and warnings. Status messages show
// up in the `Info` column of the progress display, but not in the final output. Messages can
// have a resource URN associated with them. If no urn is provided, the message is global.
LogStatus(sev diag.Severity, urn resource.URN, msg string, streamID int32)
// Analyzer fetches the analyzer with a given name, possibly lazily allocating the plugins for it. If an analyzer
// could not be found, or an error occurred while creating it, a non-nil error is returned.
Analyzer(nm tokens.QName) (Analyzer, error)
// Provider loads a new copy of the provider for a given package. If a provider for this package could not be
// found, or an error occurs while creating it, a non-nil error is returned.
Provider(pkg tokens.Package, version *semver.Version) (Provider, error)
// CloseProvider closes the given provider plugin and deregisters it from this host.
CloseProvider(provider Provider) error
// LanguageRuntime fetches the language runtime plugin for a given language, lazily allocating if necessary. If
// an implementation of this language runtime wasn't found, on an error occurs, a non-nil error is returned.
LanguageRuntime(runtime string) (LanguageRuntime, error)
// ListPlugins lists all plugins that have been loaded, with version information.
ListPlugins() []workspace.PluginInfo
// EnsurePlugins ensures all plugins in the given array are loaded and ready to use. If any plugins are missing,
// and/or there are errors loading one or more plugins, a non-nil error is returned.
EnsurePlugins(plugins []workspace.PluginInfo, kinds Flags) error
// GetRequiredPlugins lists a full set of plugins that will be required by the given program.
GetRequiredPlugins(info ProgInfo, kinds Flags) ([]workspace.PluginInfo, error)
// SignalCancellation asks all resource providers to gracefully shut down and abort any ongoing
// operations. Operation aborted in this way will return an error (e.g., `Update` and `Create`
// will either a creation error or an initialization error. SignalCancellation is advisory and
// non-blocking; it is up to the host to decide how long to wait after SignalCancellation is
// called before (e.g.) hard-closing any gRPC connection.
SignalCancellation() error
// Close reclaims any resources associated with the host.
Close() error
}
// NewDefaultHost implements the standard plugin logic, using the standard installation root to find them.
func NewDefaultHost(ctx *Context, config ConfigSource, runtimeOptions map[string]interface{}) (Host, error) {
host := &defaultHost{
ctx: ctx,
config: config,
runtimeOptions: runtimeOptions,
analyzerPlugins: make(map[tokens.QName]*analyzerPlugin),
languagePlugins: make(map[string]*languagePlugin),
resourcePlugins: make(map[Provider]*resourcePlugin),
reportedResourcePlugins: make(map[string]struct{}),
loadRequests: make(chan pluginLoadRequest),
}
// Fire up a gRPC server to listen for requests. This acts as a RPC interface that plugins can use
// to "phone home" in case there are things the host must do on behalf of the plugins (like log, etc).
svr, err := newHostServer(host, ctx)
if err != nil {
return nil, err
}
host.server = svr
// Start a goroutine we'll use to satisfy load requests serially and avoid race conditions.
go func() {
for req := range host.loadRequests {
req.result <- req.load()
}
}()
return host, nil
}
type pluginLoadRequest struct {
load func() error
result chan<- error
}
type defaultHost struct {
ctx *Context // the shared context for this host.
config ConfigSource // the source for provider configuration parameters.
runtimeOptions map[string]interface{} // options to pass to the language plugins.
analyzerPlugins map[tokens.QName]*analyzerPlugin // a cache of analyzer plugins and their processes.
languagePlugins map[string]*languagePlugin // a cache of language plugins and their processes.
resourcePlugins map[Provider]*resourcePlugin // the set of loaded resource plugins.
reportedResourcePlugins map[string]struct{} // the set of unique resource plugins we'll report.
plugins []workspace.PluginInfo // a list of plugins allocated by this host.
loadRequests chan pluginLoadRequest // a channel used to satisfy plugin load requests.
server *hostServer // the server's RPC machinery.
}
var _ Host = (*defaultHost)(nil)
type analyzerPlugin struct {
Plugin Analyzer
Info workspace.PluginInfo
}
type languagePlugin struct {
Plugin LanguageRuntime
Info workspace.PluginInfo
}
type resourcePlugin struct {
Plugin Provider
Info workspace.PluginInfo
}
func (host *defaultHost) ServerAddr() string {
return host.server.Address()
}
func (host *defaultHost) Log(sev diag.Severity, urn resource.URN, msg string, streamID int32) {
host.ctx.Diag.Logf(sev, diag.StreamMessage(urn, msg, streamID))
}
func (host *defaultHost) LogStatus(sev diag.Severity, urn resource.URN, msg string, streamID int32) {
host.ctx.StatusDiag.Logf(sev, diag.StreamMessage(urn, msg, streamID))
}
// loadPlugin sends an appropriate load request to the plugin loader and returns the loaded plugin (if any) and error.
func (host *defaultHost) loadPlugin(load func() (interface{}, error)) (interface{}, error) {
var plugin interface{}
result := make(chan error)
host.loadRequests <- pluginLoadRequest{
load: func() error {
p, err := load()
plugin = p
return err
},
result: result,
}
return plugin, <-result
}
func (host *defaultHost) Analyzer(name tokens.QName) (Analyzer, error) {
plugin, err := host.loadPlugin(func() (interface{}, error) {
// First see if we already loaded this plugin.
if plug, has := host.analyzerPlugins[name]; has {
contract.Assert(plug != nil)
return plug.Plugin, nil
}
// If not, try to load and bind to a plugin.
plug, err := NewAnalyzer(host, host.ctx, name)
if err == nil && plug != nil {
info, infoerr := plug.GetPluginInfo()
if infoerr != nil {
return nil, infoerr
}
// Memoize the result.
host.plugins = append(host.plugins, info)
host.analyzerPlugins[name] = &analyzerPlugin{Plugin: plug, Info: info}
}
return plug, err
})
if plugin == nil || err != nil {
return nil, err
}
return plugin.(Analyzer), nil
}
func (host *defaultHost) Provider(pkg tokens.Package, version *semver.Version) (Provider, error) {
plugin, err := host.loadPlugin(func() (interface{}, error) {
// Try to load and bind to a plugin.
plug, err := NewProvider(host, host.ctx, pkg, version)
if err == nil && plug != nil {
info, infoerr := plug.GetPluginInfo()
if infoerr != nil {
return nil, infoerr
}
// Warn if the plugin version was not what we expected
if version != nil && !cmdutil.IsTruthy(os.Getenv("PULUMI_DEV")) {
if info.Version == nil || !info.Version.GTE(*version) {
var v string
if info.Version != nil {
v = info.Version.String()
}
host.ctx.Diag.Warningf(
diag.Message("", /*urn*/
"resource plugin %s is expected to have version >=%s, but has %s; "+
"the wrong version may be on your path, or this may be a bug in the plugin"),
info.Name, version.String(), v)
}
}
// Record the result and add the plugin's info to our list of loaded plugins if it's the first copy of its
// kind.
key := info.Name
if info.Version != nil {
key += info.Version.String()
}
_, alreadyReported := host.reportedResourcePlugins[key]
if !alreadyReported {
host.reportedResourcePlugins[key] = struct{}{}
host.plugins = append(host.plugins, info)
}
host.resourcePlugins[plug] = &resourcePlugin{Plugin: plug, Info: info}
}
return plug, err
})
if plugin == nil || err != nil {
return nil, err
}
return plugin.(Provider), nil
}
func (host *defaultHost) LanguageRuntime(runtime string) (LanguageRuntime, error) {
plugin, err := host.loadPlugin(func() (interface{}, error) {
// First see if we already loaded this plugin.
if plug, has := host.languagePlugins[runtime]; has {
contract.Assert(plug != nil)
return plug.Plugin, nil
}
// If not, allocate a new one.
plug, err := NewLanguageRuntime(host, host.ctx, runtime, host.runtimeOptions)
if err == nil && plug != nil {
info, infoerr := plug.GetPluginInfo()
if infoerr != nil {
return nil, infoerr
}
// Memoize the result.
host.plugins = append(host.plugins, info)
host.languagePlugins[runtime] = &languagePlugin{Plugin: plug, Info: info}
}
return plug, err
})
if plugin == nil || err != nil {
return nil, err
}
return plugin.(LanguageRuntime), nil
}
func (host *defaultHost) ListPlugins() []workspace.PluginInfo {
return host.plugins
}
// EnsurePlugins ensures all plugins in the given array are loaded and ready to use. If any plugins are missing,
// and/or there are errors loading one or more plugins, a non-nil error is returned.
func (host *defaultHost) EnsurePlugins(plugins []workspace.PluginInfo, kinds Flags) error {
// Use a multieerror to track failures so we can return one big list of all failures at the end.
var result error
for _, plugin := range plugins {
switch plugin.Kind {
case workspace.AnalyzerPlugin:
if kinds&AnalyzerPlugins != 0 {
if _, err := host.Analyzer(tokens.QName(plugin.Name)); err != nil {
result = multierror.Append(result,
errors.Wrapf(err, "failed to load analyzer plugin %s", plugin.Name))
}
}
case workspace.LanguagePlugin:
if kinds&LanguagePlugins != 0 {
if _, err := host.LanguageRuntime(plugin.Name); err != nil {
result = multierror.Append(result,
errors.Wrapf(err, "failed to load language plugin %s", plugin.Name))
}
}
case workspace.ResourcePlugin:
if kinds&ResourcePlugins != 0 {
if _, err := host.Provider(tokens.Package(plugin.Name), plugin.Version); err != nil {
result = multierror.Append(result,
errors.Wrapf(err, "failed to load resource plugin %s", plugin.Name))
}
}
default:
contract.Failf("unexpected plugin kind: %s", plugin.Kind)
}
}
return result
}
// GetRequiredPlugins lists a full set of plugins that will be required by the given program.
func (host *defaultHost) GetRequiredPlugins(info ProgInfo, kinds Flags) ([]workspace.PluginInfo, error) {
var plugins []workspace.PluginInfo
if kinds&LanguagePlugins != 0 {
// First make sure the language plugin is present. We need this to load the required resource plugins.
// TODO: we need to think about how best to version this. For now, it always picks the latest.
lang, err := host.LanguageRuntime(info.Proj.Runtime.Name())
if err != nil {
return nil, errors.Wrapf(err, "failed to load language plugin %s", info.Proj.Runtime.Name())
}
plugins = append(plugins, workspace.PluginInfo{
Name: info.Proj.Runtime.Name(),
Kind: workspace.LanguagePlugin,
})
if kinds&ResourcePlugins != 0 {
// Use the language plugin to compute this project's set of plugin dependencies.
// TODO: we want to support loading precisely what the project needs, rather than doing a static scan of resolved
// packages. Doing this requires that we change our RPC interface and figure out how to configure plugins
// later than we do (right now, we do it up front, but at that point we don't know the version).
deps, err := lang.GetRequiredPlugins(info)
if err != nil {
return nil, errors.Wrapf(err, "failed to discover plugin requirements")
}
plugins = append(plugins, deps...)
}
} else {
// If we can't load the language plugin, we can't discover the resource plugins.
contract.Assertf(kinds&ResourcePlugins != 0,
"cannot load resource plugins without also loading the language plugin")
}
// Next, if there are analyzers listed in the project file, use them too.
// TODO: these are currently not versioned. We probably need to let folks specify versions in Pulumi.yaml.
if info.Proj.Analyzers != nil && kinds&AnalyzerPlugins != 0 {
for _, analyzer := range *info.Proj.Analyzers {
plugins = append(plugins, workspace.PluginInfo{
Name: string(analyzer),
Kind: workspace.AnalyzerPlugin,
})
}
}
return plugins, nil
}
func (host *defaultHost) SignalCancellation() error {
// NOTE: we're abusing loadPlugin in order to ensure proper synchronization.
_, err := host.loadPlugin(func() (interface{}, error) {
var result error
for _, plug := range host.resourcePlugins {
if err := plug.Plugin.SignalCancellation(); err != nil {
result = multierror.Append(result, errors.Wrapf(err,
"Error signaling cancellation to resource provider '%s'", plug.Info.Name))
}
}
return nil, result
})
return err
}
func (host *defaultHost) CloseProvider(provider Provider) error {
// NOTE: we're abusing loadPlugin in order to ensure proper synchronization.
_, err := host.loadPlugin(func() (interface{}, error) {
if err := provider.Close(); err != nil {
return nil, err
}
delete(host.resourcePlugins, provider)
return nil, nil
})
return err
}
func (host *defaultHost) Close() error {
// Close all plugins.
for _, plug := range host.analyzerPlugins {
if err := plug.Plugin.Close(); err != nil {
logging.Infof("Error closing '%s' analyzer plugin during shutdown; ignoring: %v", plug.Info.Name, err)
}
}
for _, plug := range host.resourcePlugins {
if err := plug.Plugin.Close(); err != nil {
logging.Infof("Error closing '%s' resource plugin during shutdown; ignoring: %v", plug.Info.Name, err)
}
}
for _, plug := range host.languagePlugins {
if err := plug.Plugin.Close(); err != nil {
logging.Infof("Error closing '%s' language plugin during shutdown; ignoring: %v", plug.Info.Name, err)
}
}
// Empty out all maps.
host.analyzerPlugins = make(map[tokens.QName]*analyzerPlugin)
host.languagePlugins = make(map[string]*languagePlugin)
host.resourcePlugins = make(map[Provider]*resourcePlugin)
// Shut down the plugin loader.
close(host.loadRequests)
// Finally, shut down the host's gRPC server.
return host.server.Cancel()
}
// Flags can be used to filter out plugins during loading that aren't necessary.
type Flags int
const (
// AnalyzerPlugins is used to only load analyzers.
AnalyzerPlugins Flags = 1 << iota
// LanguagePlugins is used to only load language plugins.
LanguagePlugins
// ResourcePlugins is used to only load resource provider plugins.
ResourcePlugins
)
// AllPlugins uses flags to ensure that all plugin kinds are loaded.
var AllPlugins = AnalyzerPlugins | LanguagePlugins | ResourcePlugins