-
Notifications
You must be signed in to change notification settings - Fork 84
/
broker.go
1599 lines (1404 loc) · 55.7 KB
/
broker.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
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
//
// Copyright (c) 2018 Red Hat, Inc.
//
// 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 broker
import (
"crypto/md5"
"encoding/hex"
"errors"
"fmt"
"reflect"
"regexp"
"strings"
"github.com/automationbroker/bundle-lib/bundle"
"github.com/automationbroker/bundle-lib/registries"
"github.com/automationbroker/config"
"github.com/openshift/ansible-service-broker/pkg/dao"
"github.com/openshift/ansible-service-broker/pkg/metrics"
"github.com/pborman/uuid"
log "github.com/sirupsen/logrus"
)
var (
// ErrorAlreadyProvisioned - Error for when an service instance has already been provisioned
ErrorAlreadyProvisioned = errors.New("already provisioned")
// ErrorDuplicate - Error for when a duplicate service instance already exists
ErrorDuplicate = errors.New("duplicate instance")
// ErrorNotFound - Error for when a service instance is not found. (either etcd or kubernetes)
ErrorNotFound = errors.New("not found")
// ErrorBindingExists - Error for when deprovision is called on a service
// instance with active bindings, or bind requested for already-existing
// binding
ErrorBindingExists = errors.New("binding exists")
// ErrorProvisionInProgress - Error for when provision is called on a service instance that has a provision job in progress
ErrorProvisionInProgress = errors.New("provision in progress")
// ErrorDeprovisionInProgress - Error for when deprovision is called on a service instance that has a deprovision job in progress
ErrorDeprovisionInProgress = errors.New("deprovision in progress")
// ErrorUpdateInProgress - Error for when update is called on a service instance that has an update job in progress
ErrorUpdateInProgress = errors.New("update in progress")
// ErrorPlanNotFound - Error for when plan for update not found
ErrorPlanNotFound = errors.New("plan not found")
// ErrorParameterNotUpdatable - Error for when parameter in update request is not updatable
ErrorParameterNotUpdatable = errors.New("parameter not updatable")
// ErrorParameterNotFound - Error for when a parameter for update is not found
ErrorParameterNotFound = errors.New("parameter not found")
// ErrorParameterUnknownEnum - Error for when an unknown enum param has been requested
ErrorParameterUnknownEnum = errors.New("unknown enum parameter value requested")
// ErrorPlanUpdateNotPossible - Error when a Plan Update request cannot be satisfied
ErrorPlanUpdateNotPossible = errors.New("plan update not possible")
// ErrorUnbindingInProgress - Error when unbind is called that has an unbinding job in progress
ErrorUnbindingInProgress = errors.New("unbinding in progress")
)
const (
// fqNameRegex - regular expression used when forming FQName.
fqNameRegex = "[/.:-]"
)
// Broker - A broker is used to to complete all the tasks that a broker must be able to do.
type Broker interface {
Bootstrap() (*BootstrapResponse, error)
Catalog() (*CatalogResponse, error)
Provision(uuid.UUID, *ProvisionRequest, bool, UserInfo) (*ProvisionResponse, error)
Update(uuid.UUID, *UpdateRequest, bool, UserInfo) (*UpdateResponse, error)
Deprovision(bundle.ServiceInstance, string, bool, bool, UserInfo) (*DeprovisionResponse, error)
Bind(bundle.ServiceInstance, uuid.UUID, *BindRequest, bool, UserInfo) (*BindResponse, bool, error)
Unbind(bundle.ServiceInstance, bundle.BindInstance, string, bool, bool, UserInfo) (*UnbindResponse, bool, error)
LastOperation(uuid.UUID, *LastOperationRequest) (*LastOperationResponse, error)
Recover() (string, error)
GetServiceInstance(uuid.UUID) (bundle.ServiceInstance, error)
GetBindInstance(uuid.UUID) (bundle.BindInstance, error)
GetBind(bundle.ServiceInstance, uuid.UUID) (*BindResponse, error)
}
// Config - Configuration for the broker.
type Config struct {
DevBroker bool `yaml:"dev_broker"`
LaunchApbOnBind bool `yaml:"launch_apb_on_bind"`
BootstrapOnStartup bool `yaml:"bootstrap_on_startup"`
Recovery bool `yaml:"recovery"`
OutputRequest bool `yaml:"output_request"`
SSLCertKey string `yaml:"ssl_cert_key"`
SSLCert string `yaml:"ssl_cert"`
RefreshInterval string `yaml:"refresh_interval"`
AutoEscalate bool `yaml:"auto_escalate"`
DashboardRedirector string `yaml:"dashboard_redirector"`
}
// DevBroker - Interface for the development broker.
type DevBroker interface {
AddSpec(spec bundle.Spec) (*CatalogResponse, error)
RemoveSpec(specID string) error
RemoveSpecs() error
}
// AnsibleBroker - Broker using ansible and images to interact with oc/kubernetes/etcd
type AnsibleBroker struct {
dao dao.Dao
registry []registries.Registry
engine *WorkEngine
brokerConfig Config
namespace string
workFactory WorkFactory
}
// NewAnsibleBroker - Creates a new ansible broker
func NewAnsibleBroker(dao dao.Dao,
registry []registries.Registry,
engine WorkEngine,
brokerConfig *config.Config,
namespace string,
workFactory WorkFactory) (*AnsibleBroker, error) {
broker := &AnsibleBroker{
dao: dao,
registry: registry,
engine: &engine,
brokerConfig: Config{
DevBroker: brokerConfig.GetBool("dev_broker"),
LaunchApbOnBind: brokerConfig.GetBool("launch_apb_on_bind"),
BootstrapOnStartup: brokerConfig.GetBool("bootstrap_on_startup"),
Recovery: brokerConfig.GetBool("recovery"),
OutputRequest: brokerConfig.GetBool("output_request"),
SSLCertKey: brokerConfig.GetString("ssl_cert_key"),
SSLCert: brokerConfig.GetString("ssl_cert"),
RefreshInterval: brokerConfig.GetString("refresh_interval"),
AutoEscalate: brokerConfig.GetBool("auto_escalate"),
DashboardRedirector: brokerConfig.GetString("dashboard_redirector"),
},
namespace: namespace,
workFactory: workFactory,
}
return broker, nil
}
// GetServiceInstance - retrieve the service instance for a instanceID.
func (a AnsibleBroker) GetServiceInstance(instanceUUID uuid.UUID) (bundle.ServiceInstance, error) {
instance, err := a.dao.GetServiceInstance(instanceUUID.String())
if err != nil {
if a.dao.IsNotFoundError(err) {
log.Infof("Could not find a service instance in dao - %v", err)
return bundle.ServiceInstance{}, ErrorNotFound
}
log.Info("Couldn't find a service instance: ", err)
return bundle.ServiceInstance{}, err
}
dashboardURL := a.getDashboardURL(instance)
if dashboardURL != "" {
instance.DashboardURL = dashboardURL
}
return *instance, nil
}
// GetBindInstance - retrieve the bind instance for a bindUUID
func (a AnsibleBroker) GetBindInstance(bindUUID uuid.UUID) (bundle.BindInstance, error) {
instance, err := a.dao.GetBindInstance(bindUUID.String())
if err != nil {
if a.dao.IsNotFoundError(err) {
return bundle.BindInstance{}, ErrorNotFound
}
return bundle.BindInstance{}, err
}
return *instance, nil
}
// Bootstrap - Loads all known specs from a registry into local storage for reference
// Potentially a large download; on the order of 10s of thousands
// TODO: How do we handle a large amount of data on this side as well? Pagination?
func (a AnsibleBroker) Bootstrap() (*BootstrapResponse, error) {
log.Info("AnsibleBroker::Bootstrap")
var err error
var specs []*bundle.Spec
var imageCount int
// Get specs from datastore
dir := "/spec"
specs, err = a.dao.BatchGetSpecs(dir)
if err != nil {
log.Errorf("Something went real bad trying to retrieve batch specs... - %v", err)
return nil, err
}
// Get list of marked specs in datastore
markedSpecs := getMarkedSpecs(specs)
// Get list of specs safe to delete
unwantedSpecs := getSafeToDeleteSpecs(a, markedSpecs)
// Delete the unwanted specs
if err := a.dao.BatchDeleteSpecs(unwantedSpecs); err != nil {
log.Errorf("Something went real bad trying to delete batch specs... - %v", err)
return nil, err
}
log.Infof("%v specs deleted", len(unwantedSpecs))
metrics.SpecsDeleted(len(unwantedSpecs))
// Getting specs again so that deleted specs do not end up in further comparisons
specs, err = a.dao.BatchGetSpecs(dir)
if err != nil {
log.Errorf("Something went real bad trying to retrieve batch specs... - %v", err)
return nil, err
}
daoSpecs := convertSpecListToMap(specs)
specs = []*bundle.Spec{}
// Load Specs for each registry
registryErrors := []error{}
for _, r := range a.registry {
s, count, err := r.LoadSpecs()
if err != nil && r.Fail(err) {
log.Errorf("registry caused bootstrap failure - %v", err)
return nil, err
}
if err != nil {
log.Warningf("registry: %v was unable to complete bootstrap - %v",
r.RegistryName(), err)
registryErrors = append(registryErrors, err)
}
imageCount += count
// this will also update the plan id
addNameAndIDForSpec(s, r.RegistryName())
specs = append(specs, s...)
metrics.SpecsLoaded(r.RegistryName(), len(s))
}
if len(registryErrors) == len(a.registry) {
return nil, errors.New("all registries failed on bootstrap")
}
specManifest := getSpecManifest(daoSpecs, specs)
markedSpecs = markSpecsForDeletion(daoSpecs, specManifest)
metrics.SpecsMarkedForDeletion(len(markedSpecs))
// Update specs in data-store
if err := a.dao.BatchSetSpecs(specManifest); err != nil {
return nil, err
}
// Update specs that are marked for deletion
if err := a.dao.BatchSetSpecs(markedSpecs); err != nil {
return nil, err
}
bundle.AddSecrets(specs)
return &BootstrapResponse{SpecCount: len(specs), ImageCount: imageCount}, nil
}
func markSpecsForDeletion(daoSpecs map[string]*bundle.Spec, specManifest bundle.SpecManifest) map[string]*bundle.Spec {
markedSpecs := make(map[string]*bundle.Spec)
for specID, spec := range daoSpecs {
if _, ok := specManifest[specID]; !ok {
log.Infof("spec '%v' marked for deletion", specID)
spec.Delete = true
markedSpecs[specID] = spec
}
}
return markedSpecs
}
func getSpecManifest(daoSpecs map[string]*bundle.Spec, specs []*bundle.Spec) bundle.SpecManifest {
specManifest := make(map[string]*bundle.Spec)
for _, s := range specs {
// If condition is just for logging. It is useful information
// as to which specs were added and which were updated
if _, ok := daoSpecs[s.ID]; ok {
log.Debugf("spec '%v|%v' needs to be updated", s.ID, s.FQName)
specManifest[s.ID] = s
s.Delete = false
} else {
log.Debugf("spec '%v|%v' needs to be added", s.ID, s.FQName)
specManifest[s.ID] = s
s.Delete = false
}
// each of the plans from all of the specs gets its own uuid. even
// though the names may be the same we want them to be globally unique.
for _, p := range s.Plans {
if p.ID == "" {
log.Errorf("We have a plan that did not get its id generated: %v", p.Name)
continue
}
}
}
return specManifest
}
// getSafeToDeleteSpecs - will return a list of specs that are safe to delete.
func getSafeToDeleteSpecs(a AnsibleBroker, markedSpecs map[string]*bundle.Spec) []*bundle.Spec {
safeToDeleteSpecs := make([]*bundle.Spec, 0)
bundleInstances, err := a.dao.BatchGetBundleInstances()
if err != nil {
log.Errorf("error getting bundle instances '%+v'", err)
// returning nil instead of checking the error,
// the broker can simply ignore to delete the specs and hope
// that the error won't repeat during next bootstrap cycle
return nil
}
log.Debugf("markedSpecs: %+v\n", markedSpecs)
for _, bundleInstance := range bundleInstances {
log.Debugf("bundle instance: %+v\n", bundleInstance.ID)
if _, ok := markedSpecs[bundleInstance.Spec.ID]; ok {
log.Debugf("spec '%v' not safe to delete", bundleInstance.Spec.ID)
delete(markedSpecs, bundleInstance.Spec.ID)
if len(markedSpecs) == 0 {
break
}
}
}
for _, spec := range markedSpecs {
log.Debugf("spec '%v' safe to delete", spec.ID)
safeToDeleteSpecs = append(safeToDeleteSpecs, spec)
}
return safeToDeleteSpecs
}
// convertSpecListToMap - will convert list of specs into a map, mapping
// specID to actual spec.
func convertSpecListToMap(specs []*bundle.Spec) map[string]*bundle.Spec {
specMap := make(map[string]*bundle.Spec)
for _, spec := range specs {
specMap[spec.ID] = spec
}
log.Debugf("converting []*Spec to SpecManifest successful")
return specMap
}
func getMarkedSpecs(specs []*bundle.Spec) map[string]*bundle.Spec {
markedSpecs := make(map[string]*bundle.Spec)
for _, spec := range specs {
if spec.Delete {
log.Debugf("spec '%v' marked for deletion", spec.ID)
markedSpecs[spec.ID] = spec
}
}
return markedSpecs
}
// addNameAndIDForSpec - will create the unique spec name and id
// and set it for each spec
func addNameAndIDForSpec(specs []*bundle.Spec, registryName string) {
for _, spec := range specs {
// need to make / a hyphen to allow for global uniqueness
// but still match spec.
re := regexp.MustCompile(fqNameRegex)
spec.FQName = re.ReplaceAllLiteralString(
fmt.Sprintf("%v-%v", registryName, spec.FQName),
"-")
spec.FQName = fmt.Sprintf("%.51v", spec.FQName)
if strings.HasSuffix(spec.FQName, "-") {
spec.FQName = spec.FQName[:len(spec.FQName)-1]
}
// ID Will be a md5 hash of the fully qualified spec name.
hasher := md5.New()
hasher.Write([]byte(spec.FQName))
spec.ID = hex.EncodeToString(hasher.Sum(nil))
// update the id on the plans, doing it here avoids looping through the
// specs array again
addIDForPlan(spec.Plans, spec.FQName)
}
}
// addIDForPlan - for each of the plans create a new ID
func addIDForPlan(plans []bundle.Plan, FQSpecName string) {
// need to use the index into the array to actually update the struct.
for i, plan := range plans {
//plans[i].ID = uuid.New()
FQPlanName := fmt.Sprintf("%s-%s", FQSpecName, plan.Name)
hasher := md5.New()
hasher.Write([]byte(FQPlanName))
plans[i].ID = hex.EncodeToString(hasher.Sum(nil))
}
}
// Recover - Will recover the broker.
func (a AnsibleBroker) Recover() (string, error) {
var emptyToken string
// do we have any jobs that are still running?
// get all in progress jobs
// For each job, check the status of each of their containers to update
// their status in case any of them finished.
recoverStatuses, err := a.dao.FindJobStateByState(bundle.StateInProgress)
if err != nil {
// no jobs or states to recover, this is OK.
if a.dao.IsNotFoundError(err) {
log.Info("No jobs to recover")
return "", nil
}
return emptyToken, err
}
/*
if a job was in progress, we know the instanceuuid & token.
do we have a podname?
if no, the job never started
we should restart the job
if yes,
did the job finish?
yes
* update status to finished
* extractCreds if available
no
* create a monitoring job to update status
*/
// let's see if we need to recover any of these
for _, rs := range recoverStatuses {
// We have an in progress job
instanceID := rs.InstanceID.String()
instance, err := a.dao.GetServiceInstance(instanceID)
if err != nil {
return emptyToken, err
}
// Do we have a podname?
if rs.State.Podname == "" {
// NO, we do not have a podname
log.Infof("No podname. Attempting to restart job: %s", instanceID)
log.Debugf("%v", instance)
// Handle bad write of service instance
if instance.Spec == nil || instance.Parameters == nil {
a.dao.SetState(instanceID, bundle.JobState{
Token: rs.State.Token,
State: bundle.StateFailed,
Method: rs.State.Method,
})
a.dao.DeleteServiceInstance(instance.ID.String())
log.Warningf("incomplete ServiceInstance [%s] record, marking job as failed",
instance.ID)
// skip to the next item
continue
}
var job Work
var topic WorkTopic
if rs.State.Method == bundle.JobMethodProvision {
job = a.workFactory.NewProvisionJob(instance)
topic = ProvisionTopic
} else if rs.State.Method == bundle.JobMethodUpdate {
job = a.workFactory.NewUpdateJob(instance)
topic = UpdateTopic
} else if rs.State.Method == bundle.JobMethodDeprovision {
job = a.workFactory.NewDeprovisionJob(instance, false)
topic = DeprovisionTopic
} else {
log.Warningf(
"Attempted to recover job %s, but found an unrecognized "+
"MethodType: %s, skipping...",
rs.State.Token, rs.State.Method,
)
}
// Need to use the same token as before, since that's what the
// catalog will try to ping.
_, err := a.engine.StartNewAsyncJob(rs.State.Token, job, topic)
if err != nil {
return emptyToken, err
}
} else {
// YES, we have a podname
log.Infof("We have a pod to recover: %s", rs.State.Podname)
// did the pod finish?
extErr := bundle.RecoverExtractCredentials(
rs.State.Podname,
instance.Context.Namespace,
instance.Spec.FQName,
instance.ID.String(),
rs.State.Method,
[]string{instance.Context.Namespace},
instance.Spec.Runtime,
)
// NO, pod failed.
if extErr != nil {
log.Errorf("broker::Recover error occurred. %s", extErr.Error())
return emptyToken, extErr
}
}
}
log.Info("Recovery complete")
return "recover called", nil
}
// Catalog - returns the catalog of services defined
func (a AnsibleBroker) Catalog() (*CatalogResponse, error) {
log.Info("AnsibleBroker::Catalog")
var specs []*bundle.Spec
var err error
var services []Service
dir := "/spec"
if specs, err = a.dao.BatchGetSpecs(dir); err != nil {
log.Error("Something went real bad trying to retrieve batch specs...")
return nil, err
}
log.Debugf("Filtering secret parameters out of specs...")
specs, err = bundle.FilterSecrets(specs)
if err != nil {
// Should we blow up or warn and continue?
log.Errorf("Something went real bad trying to load secrets %v", err)
return nil, err
}
services = []Service{}
for _, spec := range specs {
ser, err := SpecToService(spec)
if err != nil {
log.Errorf("not adding spec %v to list of services due to error transforming to service - %v", spec.FQName, err)
} else {
// Bug 1539542 - in order for async bind to work,
// bindings_retrievable needs to be set to true. We only want to
// set BindingsRetrievable to true if the service is bindable
// AND we the broker is configured to launch apbs on bind
if ser.Bindable && a.brokerConfig.LaunchApbOnBind {
ser.BindingsRetrievable = true
}
if !spec.Delete {
// add only the specs that are not marked for deletion.
services = append(services, ser)
}
}
}
return &CatalogResponse{services}, nil
}
// Provision - will provision a service
func (a AnsibleBroker) Provision(instanceUUID uuid.UUID, req *ProvisionRequest, async bool, userInfo UserInfo,
) (*ProvisionResponse, error) {
////////////////////////////////////////////////////////////
//type ProvisionRequest struct {
//-> OrganizationID uuid.UUID
//-> SpaceID uuid.UUID
// Used for determining where this service should be provisioned. Analogous to
// OCP's namespaces and projects. Re: OrganizationID, spec mentions
// "Most brokers will not use this field, it could be helpful in determining
// the data placement or applying custom business rules"
//-> PlanID uuid.UUID
//-> ServiceID uuid.UUID
// ServiceID maps directly to a Spec.Id found in etcd. Can pull Spec via
// Dao::GetSpec(id string)
//-> Parameters map[string]string
// User provided configuration answers for the AnsibleApp
// -> AcceptsIncomplete bool
// true indicates both the SC and the requesting client (sc client). If param
// is not included in the req, and the broker can only provision an instance of
// the request plan asynchronously, broker should reject with a 422
// NOTE: Spec.Async should indicate what level of async support is available for
// a given ansible app
//}
// Summary:
// For our purposes right now, the ServiceID and the Params should be enough to
// Provision an ansible app.
////////////////////////////////////////////////////////////
// Provision Flow
// -> Retrieve Spec from etcd (if missing, 400, this returns err missing)
// -> Make entry in /instance, ID'd by instance. Value should be Instance type
// Purpose is to make sure everything need to deprovision is available
// in persistence.
// -> Provision!
////////////////////////////////////////////////////////////
/*
dao GET returns error strings like CODE: message (entity) [#]
dao SetServiceInstance returns what error?
dao.SetState returns what error?
Provision returns what error?
SetExtractedCredentials returns what error?
broker
* normal synchronous return ProvisionResponse
* normal async return ProvisionResponse
* if instance already exists with the same params, return ProvisionResponse, AND InstanceExists
* if instance already exists DIFFERENT param, return nil AND InstanceExists
handler returns the following
* synchronous provision return 201 created
* instance already exists with IDENTICAL parameters to existing instance, 200 OK
* async provision 202 Accepted
* instance already exists with DIFFERENT parameters, 409 Conflict {}
* if only support async and no accepts_incomplete=true passed in, 422 Unprocessable entity
*/
var spec *bundle.Spec
var err error
// Retrieve requested spec
specID := req.ServiceID
if spec, err = a.dao.GetSpec(specID); err != nil {
// etcd return not found i.e. code 100
if a.dao.IsNotFoundError(err) {
return nil, ErrorNotFound
}
// otherwise unknown error bubble it up
return nil, err
}
// If the spec is marked for Deletion, send the same
// error as Spec Not Found
if spec.Delete {
return nil, ErrorNotFound
}
context := &req.Context
parameters := req.Parameters
if parameters == nil {
parameters = make(bundle.Parameters)
}
if req.PlanID == "" {
errMsg :=
"PlanID from provision request is blank. " +
"Provision requests must specify PlanIDs"
return nil, errors.New(errMsg)
}
plan, ok := spec.GetPlanFromID(req.PlanID)
if !ok {
return nil, ErrorNotFound
}
log.Debugf(
"Injecting PlanID as parameter: { %s: %s }",
planParameterKey, plan.Name)
parameters[planParameterKey] = plan.Name
log.Debugf("Injecting ServiceClassID as parameter: { %s: %s }",
serviceClassIDKey, req.ServiceID)
parameters[serviceClassIDKey] = req.ServiceID
log.Debugf("Injecting ServiceInstanceID as parameter: { %s: %s }",
serviceInstIDKey, instanceUUID.String())
parameters[serviceInstIDKey] = instanceUUID.String()
log.Debugf("Injecting lastRequestingUserKey as parameter: { %s: %s }",
lastRequestingUserKey, getLastRequestingUser(userInfo))
parameters[lastRequestingUserKey] = getLastRequestingUser(userInfo)
// Build and persist record of service instance
serviceInstance := &bundle.ServiceInstance{
ID: instanceUUID,
Spec: spec,
Context: context,
Parameters: ¶meters,
}
// Verify we're not re-provisioning the same instance
// if err is nil, there is an instance. Let's compare it to the instance
// we're being asked to provision.
//
// if err is not nil, we will just bubble that up
si, err := a.dao.GetServiceInstance(instanceUUID.String())
if err != nil && !a.dao.IsNotFoundError(err) {
return nil, err
}
// This will use the package to make sure that if the type is changed
// away from []byte it can still be evaluated.
if si != nil && uuid.Equal(si.ID, serviceInstance.ID) {
if reflect.DeepEqual(si.Parameters, serviceInstance.Parameters) {
alreadyInProgress, jobToken, err := a.isJobInProgress(serviceInstance.ID.String(), bundle.JobMethodProvision)
if err != nil {
return nil, fmt.Errorf("An error occurred while trying to determine if a provision job is already in progress for instance: %s", serviceInstance.ID)
}
if alreadyInProgress {
log.Infof("Provision requested for instance %s, but job is already in progress", serviceInstance.ID)
return &ProvisionResponse{Operation: jobToken}, ErrorProvisionInProgress
}
log.Debug("already have this instance returning 200")
return &ProvisionResponse{}, ErrorAlreadyProvisioned
}
log.Info("we have a duplicate instance with parameters that differ, returning 409 conflict")
return nil, ErrorDuplicate
}
//
// Looks like this is a new provision, let's get started.
//
if err = a.dao.SetServiceInstance(instanceUUID.String(), serviceInstance); err != nil {
return nil, err
}
var token = a.engine.Token()
pjob := a.workFactory.NewProvisionJob(serviceInstance)
metrics.ActionStarted("provision")
if async {
log.Info("ASYNC provisioning in progress")
// asynchronously provision and return the token for the lastoperation
token, err = a.engine.StartNewAsyncJob(token, pjob, ProvisionTopic)
if err != nil {
log.Errorf("Failed to start new job for async provision\n%s", err.Error())
return nil, err
}
} else {
log.Info("reverting to synchronous provisioning in progress")
if err := a.engine.StartNewSyncJob(token, pjob, ProvisionTopic); err != nil {
log.Errorf("Failed to start new job for sync provision\n%s", err.Error())
return nil, err
}
}
var response *ProvisionResponse
dashboardURL := a.getDashboardURL(serviceInstance)
if dashboardURL != "" {
response = &ProvisionResponse{Operation: token, DashboardURL: dashboardURL}
} else {
response = &ProvisionResponse{Operation: token}
}
return response, nil
}
// getDashboardURL - will conditionally return a dashboard redirector url or
// an empty string if the redirector feature is not specified by the APB.
func (a *AnsibleBroker) getDashboardURL(si *bundle.ServiceInstance) string {
var val interface{}
var drEnabled, ok bool
spec := si.Spec
if len(spec.Alpha) == 0 {
return ""
}
val, ok = spec.Alpha["dashboard_redirect"]
if !ok {
return ""
}
drEnabled, ok = val.(bool)
if !ok {
return ""
}
if !drEnabled {
return ""
}
if a.brokerConfig.DashboardRedirector == "" {
log.Warningf("Attempting to provision %v, which has dashboard redirect enabled, "+
"but no dashboard_redirector route was found in the broker's configmap. "+
"Deploying without a dashboard_url.", spec.FQName)
return ""
}
drURL := a.brokerConfig.DashboardRedirector
if !strings.HasPrefix(drURL, "http") {
drURL = fmt.Sprintf("http://%s", drURL)
}
return fmt.Sprintf("%s/?id=%s", drURL, si.ID)
}
// Deprovision - will deprovision a service.
func (a AnsibleBroker) Deprovision(
instance bundle.ServiceInstance, planID string, skipApbExecution bool, async bool, userInfo UserInfo,
) (*DeprovisionResponse, error) {
////////////////////////////////////////////////////////////
// Deprovision flow
// -> Lookup bindings by instance ID; 400 if any are active, related issue:
// https://github.com/openservicebrokerapi/servicebroker/issues/127
// -> Atomic deprovision and removal of service entry in etcd?
// * broker::Deprovision
// Arguments for this? What data do apbs require to deprovision?
// * namespace
// Maybe just hand off a serialized ServiceInstance and let the apb
// decide what's important?
// * delete credentials from etcd
// * if noerror: delete serviceInstance entry with Dao
if planID == "" {
errMsg := "Deprovision request contains an empty plan_id"
return nil, errors.New(errMsg)
}
err := a.validateDeprovision(&instance)
if err != nil {
return nil, err
}
alreadyInProgress, jobToken, err := a.isJobInProgress(instance.ID.String(), bundle.JobMethodDeprovision)
if err != nil {
return nil, fmt.Errorf("An error occurred while trying to determine if a deprovision job is already in progress for instance: %s", instance.ID)
}
if alreadyInProgress {
log.Infof("Deprovision requested for instance %s, but job is already in progress", instance.ID)
return &DeprovisionResponse{Operation: jobToken}, ErrorDeprovisionInProgress
}
provExtCreds, err := bundle.GetExtractedCredentials(instance.ID.String())
if err != nil && err != bundle.ErrExtractedCredentialsNotFound {
log.Warningf("unable to retrieve provision time credentials - %v", err)
return nil, err
}
// Add the DB Credentials to the parameters. This will allow the apb to use these credentials
// if it so chooses.
if provExtCreds != nil && instance.Parameters != nil {
params := *instance.Parameters
params[bundle.ProvisionCredentialsKey] = provExtCreds.Credentials
instance.Parameters = ¶ms
}
// Override the lastRequestingUserKey value in the instance.Parameters
if instance.Parameters != nil {
(*instance.Parameters)[lastRequestingUserKey] = getLastRequestingUser(userInfo)
instance.Parameters.EnsureDefaults()
}
var token = a.engine.Token()
dpjob := a.workFactory.NewDeprovisionJob(&instance, skipApbExecution)
metrics.ActionStarted("deprovision")
if async {
log.Info("ASYNC deprovision in progress")
token, err = a.engine.StartNewAsyncJob(token, dpjob, DeprovisionTopic)
if err != nil {
log.Errorf("Failed to start new job for async deprovision\n%s", err.Error())
return nil, err
}
return &DeprovisionResponse{Operation: token}, nil
}
if !skipApbExecution {
log.Info("Synchronous deprovision in progress")
if err := a.engine.StartNewSyncJob(token, dpjob, DeprovisionTopic); err != nil {
return nil, err
}
// Attempt to delete extracted credentials created from provision. The creds
// should only be deleted if the instance was successfully deleted, otherwise
// we will end up in a corrupt state.
if err := bundle.DeleteExtractedCredentials(instance.ID.String()); err != nil {
log.Infof("Attempted to delete extracted credentials from a provision but could not: %s", err.Error())
}
}
return &DeprovisionResponse{}, nil
}
func (a AnsibleBroker) validateDeprovision(instance *bundle.ServiceInstance) error {
// -> Lookup bindings by instance ID; 400 if any are active, related issue:
// https://github.com/openservicebrokerapi/servicebroker/issues/127
if len(instance.BindingIDs) > 0 {
log.Debugf("Found bindings with ids: %v", instance.BindingIDs)
return ErrorBindingExists
}
return nil
}
func (a AnsibleBroker) isJobInProgress(ID string,
method bundle.JobMethod) (bool, string, error) {
allJobs, err := a.dao.GetSvcInstJobsByState(ID, bundle.StateInProgress)
log.Infof("All Jobs for instance: %v in state: %v - \n%#v", ID, bundle.StateInProgress, allJobs)
if err != nil {
return false, "", err
}
var token string
methodJobs := dao.MapJobStatesWithMethod(allJobs, method)
if len(methodJobs) > 0 {
token = methodJobs[0].Token
}
return len(methodJobs) > 0, token, nil
}
// GetBind - will return the binding between a service created via an async
// binding event.
func (a AnsibleBroker) GetBind(instance bundle.ServiceInstance, bindingUUID uuid.UUID) (*BindResponse, error) {
log.Debug("broker.GetBind: entered GetBind")
provExtCreds, err := bundle.GetExtractedCredentials(instance.ID.String())
if err != nil && err != bundle.ErrExtractedCredentialsNotFound {
log.Warningf("unable to retrieve provision time credentials - %v", err)
return nil, err
}
bi, err := a.dao.GetBindInstance(bindingUUID.String())
if err != nil {
if a.dao.IsNotFoundError(err) {
log.Warningf("id: %v - could not find bind instance - %v", bindingUUID, err)
return nil, ErrorNotFound
}
log.Warningf("id: %v - unable to retrieve bind instance - %v", bindingUUID, err)
return nil, err
}
bindExtCreds, err := bundle.GetExtractedCredentials(bi.ID.String())
if err != nil {
if err == bundle.ErrExtractedCredentialsNotFound {
return nil, ErrorNotFound
}
return nil, err
}
log.Debug("broker.GetBind: we got the bind credentials")
return NewBindResponse(provExtCreds, bindExtCreds)
}
// Bind - will create a binding between a service. Parameter "async" declares
// whether the caller is willing to have the operation run asynchronously. The
// returned bool will be true if the operation actually ran asynchronously.
func (a AnsibleBroker) Bind(instance bundle.ServiceInstance, bindingUUID uuid.UUID, req *BindRequest, async bool, userInfo UserInfo,
) (*BindResponse, bool, error) {
// binding_id is the id of the binding.
// the instanceUUID is the previously provisioned service id.
//
// See if the service instance still exists, if not send back a badrequest.
// GET SERVICE get provision parameters
params := req.Parameters
if params == nil {
params = make(bundle.Parameters)
}
// Inject PlanID into parameters passed to APBs
if req.PlanID == "" {
errMsg :=
"PlanID from bind request is blank. " +
"Bind requests must specify PlanIDs"
return nil, false, errors.New(errMsg)
}
plan, ok := instance.Spec.GetPlanFromID(req.PlanID)
if !ok {
log.Debug("Plan not found")
return nil, false, ErrorNotFound
}
log.Debugf(
"Injecting PlanID as parameter: { %s: %s }",
planParameterKey, plan.Name)
params[planParameterKey] = plan.Name
log.Debugf("Injecting ServiceClassID as parameter: { %s: %s }",
serviceClassIDKey, req.ServiceID)
params[serviceClassIDKey] = req.ServiceID
log.Debugf("Injecting ServiceInstanceID as parameter: { %s: %s }",
serviceInstIDKey, instance.ID.String())
params[serviceInstIDKey] = instance.ID.String()
log.Debugf("Injecting lastRequestingUserKey as parameter: { %s: %s }",
lastRequestingUserKey, getLastRequestingUser(userInfo))
params[lastRequestingUserKey] = getLastRequestingUser(userInfo)
log.Debugf("Injecting ServiceBindingID as parameter: { %s: %s }",
serviceBindingIDKey, bindingUUID.String())
params[serviceBindingIDKey] = bindingUUID.String()
// Create a BindingInstance with a reference to the serviceinstance.
bindingInstance := &bundle.BindInstance{
ID: bindingUUID,
ServiceID: instance.ID,
Parameters: ¶ms,
}
// Verify we're not rebinding the same instance. if err is nil, there is an
// instance. Let's compare it to the instance we're being asked to bind.
//
// if err is not nil, we will just bubble that up