Crossplane 入门

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

apiVersion: database.aws.crossplane.io/v1beta1
kind: RDSInstance
metadata:
  name: rdspostgresql
spec:
  forProvider:
    region: us-east-1
    dbInstanceClass: db.t2.small
    masterUsername: masteruser
    allocatedStorage: 20
    engine: postgres
    engineVersion: "12"
    skipFinalSnapshotBeforeDeletion: true
  writeConnectionSecretToRef:
    namespace: crossplane-system
    name: aws-rdspostgresql-conn

1
2
3
4
5
6
7
8

apiVersion: database.gcp.crossplane.io/v1beta1
kind: CloudSQLInstance
metadata:
  name: foodb
  annotations:
    crossplane.io/external-name: my-special-db
spec:
# ...

1
2
3
4
5
6
7
8

spec:
  forProvider:
    resourceGroupName: foo-res-group
    resourceGroupNameRef:
      name: resourcegroup
    resourceGroupNameSelector:
      matchLabels:
        app: prod

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

apiVersion: compute.gcp.crossplane.io/v1beta1
kind: Network
metadata:
  name: foo-network
  annotations:
    crossplane.io/external-name: existing-network
spec:
  forProvider: {}
  providerConfigRef:
    name: default

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

apiVersion: database.example.org/v1alpha1
kind: XPostgreSQLInstance
metadata:
  name: my-db
spec:
  parameters:
    storageGB: 20
  compositionRef:
    name: production
  writeConnectionSecretToRef:
    namespace: crossplane-system
    name: my-db-connection-details

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

apiVersion: database.example.org/v1alpha1
kind: PostgreSQLInstance
metadata:
  namespace: default
  name: my-db
spec:
  parameters:
    storageGB: 20
  compositionRef:
    name: production
  writeConnectionSecretToRef:
    name: my-db-connection-details

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# ...
spec:
  compositionRef:
    name: production-us-east
  compositionSelector:
    matchLabels:
      environment: production
      region: us-east
      provider: gcp
# ...

 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

apiVersion: apiextensions.crossplane.io/v1
kind: CompositeResourceDefinition
metadata:
  name: xpostgresqlinstances.database.example.org
spec:
  group: database.example.org
  names:
    kind: XPostgreSQLInstance
    plural: xpostgresqlinstances
  claimNames:
    kind: PostgreSQLInstance
    plural: postgresqlinstances
  versions:
  - name: v1alpha1
    served: true
    referenceable: true
    schema:
      openAPIV3Schema:
        type: object
        properties:
          spec:
            type: object
            properties:
              parameters:
                type: object
                properties:
                  storageGB:
                    type: integer
                required:
                - storageGB
            required:
            - parameters

 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

apiVersion: apiextensions.crossplane.io/v1
kind: Composition
metadata:
  name: example
  labels:
    crossplane.io/xrd: xpostgresqlinstances.database.example.org
    provider: gcp
spec:
  writeConnectionSecretsToNamespace: crossplane-system
  compositeTypeRef:
    apiVersion: database.example.org/v1alpha1
    kind: XPostgreSQLInstance
  resources:
  - name: cloudsqlinstance
    base:
      apiVersion: database.gcp.crossplane.io/v1beta1
      kind: CloudSQLInstance
      spec:
        forProvider:
          databaseVersion: POSTGRES_12
          region: us-central1
          settings:
            tier: db-custom-1-3840
            dataDiskType: PD_SSD
            ipConfiguration:
              ipv4Enabled: true
              authorizedNetworks:
                - value: "0.0.0.0/0"
    patches:
    - type: FromCompositeFieldPath
      fromFieldPath: spec.parameters.storageGB
      toFieldPath: spec.forProvider.settings.dataDiskSizeGb

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

apiVersion: apiextensions.crossplane.io/v1
kind: Composition
spec:
  resources:
  - name: cloudsqlinstance
    base:
      # ...
    patches:
    - type: PatchSet
      patchSetName: metadata
      transforms:
      - type: map
        map:
          medium: db-custom-1-3840

1
2
3
4
5
6

apiVersion: pkg.crossplane.io/v1
kind: Provider
metadata:
  name: provider-aws
spec:
  package: "crossplane/provider-aws:master"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

apiVersion: aws.crossplane.io/v1beta1
kind: ProviderConfig
metadata:
  name: aws-provider
spec:
  credentials:
    source: Secret
    secretRef:
      namespace: crossplane-system
      name: aws-creds
      key: key

1
2
3
4
5
6
7
8

apiVersion: database.aws.crossplane.io/v1beta1
kind: RDSInstance
metadata:
  name: prod-sql
spec:
  providerConfigRef:
    name: aws-provider
  # ...

1
2
3

$ k get pods -n crossplane-system
NAME                                                            READY   STATUS    RESTARTS   AGE
crossplane-contrib-provider-aws-e0b33d016bb8-74754647cf-kltps   1/1     Running   0          5m31s

 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

type ExternalConnectDisconnecter interface {
	ExternalConnecter
	ExternalDisconnecter
}

// ExternalConnecter 创建出一个 ExternalClient，通常是创建 cloud client
type ExternalConnecter interface {
	Connect(ctx context.Context, mg resource.Managed) (ExternalClient, error)
}

// ExternalConnecter 关闭一个 ExternalClient
type ExternalDisconnecter interface {
	Disconnect(ctx context.Context) error
}

// ExternalClient 管理某个 Managed Resource 的声明周期
type ExternalClient interface {
  // Observe 查询 Managed Resource 相关的外部资源是否存在，以及更新 Status
	Observe(ctx context.Context, mg resource.Managed) (ExternalObservation, error)

  // Create 创建 Managed Resource 资源相关的外部资源
  // Create 会在 Observe 查询外部资源不存在之后调用。Create 实现中只允许更新
  // Managed Resource 的 annotation，其他的更新会被丢弃
	Create(ctx context.Context, mg resource.Managed) (ExternalCreation, error)

  // Update 更新 Managed Resource 相关的外部资源
  // Update 会在 Observe 返回需要更新之后调用
	Update(ctx context.Context, mg resource.Managed) (ExternalUpdate, error)

  // Delete 删除 Managed Resource 相关外部资源
  // Delete 在 Managed Resource 标记为删除时调用
	Delete(ctx context.Context, mg resource.Managed) error
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

func SetupBucket(mgr ctrl.Manager, l logging.Logger, rl workqueue.RateLimiter) error {
	name := managed.ControllerName(v1alpha3.BucketGroupKind)

  // 注册到 Controller Manager 中
	return ctrl.NewControllerManagedBy(mgr).
		Named(name).
		WithOptions(controller.Options{
			RateLimiter: ratelimiter.NewDefaultManagedRateLimiter(rl),
		}).
		For(&v1alpha3.Bucket{}).
		Complete(
      // 通过通用函数将 ExternalConnectDisconnecter 扩展为 Reconciler
			managed.NewReconciler(mgr,
				resource.ManagedKind(v1alpha3.BucketGroupVersionKind),
				managed.WithExternalConnecter(&connecter{client: mgr.GetClient()}),
				managed.WithLogger(l.WithValues("controller", name)),
				managed.WithRecorder(event.NewAPIRecorder(mgr.GetEventRecorderFor(name))),
			),
		)
}

  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

type Reconciler struct {
	client     client.Client            // kubernetes client
	newManaged func() resource.Managed  // factory to create a managed resource

  external mrExternal                 // ExternalConnectDisconnecter
	managed  mrManaged
  // ...
}

// Reconcile a managed resource with an external resource.
func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
	// 读取 MR CR
  managed := r.newManaged()
	if err := r.client.Get(ctx, req.NamespacedName, managed); err != nil {
		return requeue
	}

  // MR 标记为删除，并且删除策略是 Orphan，那么仅仅移除 CR Finalizer
	if meta.WasDeleted(managed) && managed.GetDeletionPolicy() == xpv1.DeletionOrphan {
		// ...
		if err := r.managed.RemoveFinalizer(ctx, managed); err != nil {
			return requeue
		}
		return norequeue
	}

  // 对 MR 做一些初始的信息填充
  // 例如设置 external name
	if err := r.managed.Initialize(ctx, managed); err != nil {
		return requeue
	}

  // 通过 MR Anno 发现创建操作处于中间态，那么不进行任何操作了
	if meta.ExternalCreateIncomplete(managed) {
		return requeue
	}

	// 解析 MR 相关依赖（填充一些字段）
  // 例如使用 spec.networkRef.name 填充 spec.network
	if !meta.WasDeleted(managed) {
		if err := r.managed.ResolveReferences(ctx, managed); err != nil {
			return requeue
		}
	}

  // 构建 ExternalClient
	external, err := r.external.Connect(externalCtx, managed)
	if err != nil {
		return requeue
	}
	defer func() {
		if err := r.external.Disconnect(ctx); err != nil {
		}
	}()

  // 执行 Observe 查询 MR 相关资源情况
	observation, err := external.Observe(externalCtx, managed)
	if err != nil {
		return requeue
	}

  // 查询结果为相关资源不存在，但是 MR Anno 表明之前创建成功，那么检查是否满足 Create 调用周期
  // 这样是因为防止一些外部资源正在创建中，但是查询可能会报告资源不存在
	if !observation.ResourceExists && meta.ExternalCreateSucceededDuring(managed, r.creationGracePeriod) {
		return requeue
	}

  // MR 标记为删除，那么执行 Delete 操作
	if meta.WasDeleted(managed) {
		if observation.ResourceExists {
			if err := external.Delete(externalCtx, managed); err != nil {
				return requeue
			}

			return requeue
		}

    // 外部资源已经删除，那么移除 MR Finalizer
		if err := r.managed.RemoveFinalizer(ctx, managed); err != nil {
			managed.SetConditions(xpv1.Deleting(), xpv1.ReconcileError(err))
			return requeue
		}

    // 完全删除，不再需要 requeue
		return norequeue
	}


  // MR 添加 Finalizer
	if err := r.managed.AddFinalizer(ctx, managed); err != nil {
		return requeue
	}

  // 外部资源不存在，走 Create/Update 逻辑
	if !observation.ResourceExists {
    // MR 打上 Pending Anno，用于判断 Create 操作是否执行完
		meta.SetExternalCreatePending(managed, time.Now())

    // 执行 Create 操作
		creation, err := external.Create(externalCtx, managed)
		if err != nil {
      // Create 失败，打上 CreateFailed Anno
			meta.SetExternalCreateFailed(managed, time.Now())
			return requeue
		}

    // Create 成功，MR 打上 CreateSucceeded Anno
		meta.SetExternalCreateSucceeded(managed, time.Now())
		return requeue
	}

  // 外部资源不需要更新，直接返回
	if observation.ResourceUpToDate {
		return requeue
	}

  // 外部资源需要更新，执行 Update 操作
	update, err := external.Update(externalCtx, managed)
	if err != nil {
		return requeue
	}
	return requeue

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

type cloudsqlConnector struct {
	kube client.Client
}

func (c *cloudsqlConnector) Connect(ctx context.Context, mg resource.Managed) (managed.ExternalClient, error) {
  // 读取 ProviderConfig 以及相关的凭证信息
	projectID, opts, err := gcp.GetAuthInfo(ctx, c.kube, mg)
	if err != nil {
		return nil, err
	}
  // 构建 GCP Client
	s, err := sqladmin.NewService(ctx, opts)
	if err != nil {
		return nil, errors.Wrap(err, errNewClient)
	}
	return &cloudsqlExternal{kube: c.kube, db: s.Instances, projectID: projectID}, nil
}

 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

type cloudsqlExternal struct {
	kube      client.Client
	db        *sqladmin.InstancesService
	projectID string
}

func (c *cloudsqlExternal) Observe(ctx context.Context, mg resource.Managed) (managed.ExternalObservation, error) {
	cr, ok := mg.(*v1beta1.CloudSQLInstance)
	if !ok {
		return managed.ExternalObservation{}, errors.New(errNotCloudSQL)
	}

  // 查看 DB 是否存在
	instance, err := c.db.Get(c.projectID, meta.GetExternalName(cr)).Context(ctx).Do()
	if err != nil {
		return managed.ExternalObservation{}, errors.Wrap(resource.Ignore(gcp.IsErrorNotFound, err), errGetFailed)
	}
	
  // 更新 MR 中的状态
	cr.Status.AtProvider = cloudsql.GenerateObservation(*instance)
	switch cr.Status.AtProvider.State {
	case v1beta1.StateRunnable:
		cr.Status.SetConditions(xpv1.Available())
	case v1beta1.StateCreating:
		cr.Status.SetConditions(xpv1.Creating())
	case v1beta1.StateCreationFailed, v1beta1.StateSuspended, v1beta1.StateMaintenance, v1beta1.StateUnknownState:
		cr.Status.SetConditions(xpv1.Unavailable())
	}

  // 对于 MR Spec 与实际的 DB 状态，判断是否需要执行 Update
	upToDate, err := cloudsql.IsUpToDate(meta.GetExternalName(cr), &cr.Spec.ForProvider, instance)
	if err != nil {
		return managed.ExternalObservation{}, errors.Wrap(err, errCheckUpToDate)
	}
	return managed.ExternalObservation{
		ResourceExists:    true,
		ResourceUpToDate:  upToDate,
		ConnectionDetails: getConnectionDetails(cr, instance),
	}, nil
}

func (c *cloudsqlExternal) Create(ctx context.Context, mg resource.Managed) (managed.ExternalCreation, error) {
	cr, ok := mg.(*v1beta1.CloudSQLInstance)
	if !ok {
		return managed.ExternalCreation{}, errors.New(errNotCloudSQL)
	}
	
  // 根据 MR Spec 执行 Create
	instance := &sqladmin.DatabaseInstance{}
	cloudsql.GenerateDatabaseInstance(meta.GetExternalName(cr), cr.Spec.ForProvider, instance)

  // GCP Client 执行创建
	if _, err := c.db.Insert(c.projectID, instance).Context(ctx).Do(); err != nil {
		if gcp.IsErrorAlreadyExists(err) {
			return managed.ExternalCreation{}, errors.Wrap(err, errNameInUse)
		}
		return managed.ExternalCreation{}, errors.Wrap(err, errCreateFailed)
	}

	cd := managed.ConnectionDetails{
		xpv1.ResourceCredentialsSecretPasswordKey: []byte(pw),
	}
	return managed.ExternalCreation{ConnectionDetails: cd}, nil
}

func (c *cloudsqlExternal) Update(ctx context.Context, mg resource.Managed) (managed.ExternalUpdate, error) {
	cr, ok := mg.(*v1beta1.CloudSQLInstance)
	if !ok {
		return managed.ExternalUpdate{}, errors.New(errNotCloudSQL)
	}
	if cr.Status.AtProvider.State == v1beta1.StateCreating {
		return managed.ExternalUpdate{}, nil
	}
  // 根据 MR Spec 执行 Patch 请求
	instance := &sqladmin.DatabaseInstance{}
	cloudsql.GenerateDatabaseInstance(meta.GetExternalName(cr), cr.Spec.ForProvider, instance)
	_, err := c.db.Patch(c.projectID, meta.GetExternalName(cr), instance).Context(ctx).Do()
	return managed.ExternalUpdate{}, errors.Wrap(err, errUpdateFailed)
}

func (c *cloudsqlExternal) Delete(ctx context.Context, mg resource.Managed) error {
	cr, ok := mg.(*v1beta1.CloudSQLInstance)
	if !ok {
		return errors.New(errNotCloudSQL)
	}
	cr.SetConditions(xpv1.Deleting())
	_, err := c.db.Delete(c.projectID, meta.GetExternalName(cr)).Context(ctx).Do()
	if gcp.IsErrorNotFound(err) {
		return nil
	}
	return errors.Wrap(err, errDeleteFailed)
}