Add source resource (streamnative#41)

* feat: add source resource
* fix: add miss dependency
* fix: improve the resources fields
* test: add acc test for pulsar_source
* fix: use online connector
* fix: test providers

Signed-off-by: Zixuan Liu <[email protected]>
Signed-off-by: Max Xu <[email protected]>

README.md

@@ -313,6 +313,57 @@ resource "pulsar_topic" "sample-topic-2" {
 | `permission_grant`            | [Permission grants](https://pulsar.apache.org/docs/en/admin-api-permissions/) on a topic. This block can be repeated for each grant you'd like to add. Permission grants are also inherited from the topic's namespace. | No |
 | `retention_policies`          | Data retention policies                                           | No |
 
+### `pulsar_source`
+
+A resource for creating and managing Apache Pulsar Sources.
+
+#### Example
+
+```hcl
+provider "pulsar" {
+  web_service_url = "http://localhost:8080"
+  api_version = "3"
+}
+
+resource "pulsar_source" "source-1" {
+  provider = pulsar
+
+  name = "source-1"
+  tenant = "public"
+  namespace = "default"
+
+  archive = "testdata/pulsar-io/pulsar-io-file-2.8.1.nar"
+
+  destination_topic_name = "source-1-topic"
+
+  processing_guarantees = "EFFECTIVELY_ONCE"
+
+  configs = "{\"inputDirectory\":\"opt\"}"
+
+  cpu = 2
+  disk_mb = 20480
+  ram_mb = 2048
+}
+```
+
+#### Properties
+
+| Property | Description | Required|  
+| ----------------------------- | ----------------------------------------------------------------- |----------------------------|
+| `name` | The source's name | True
+| `tenant` | The source's tenant | True  
+| `namespace` | The source's namespace | True
+| `destination_topic_name` | The Pulsar topic to which data is sent | True
+| `archive` | The path to the NAR archive for the Source. It also supports url-path [http/https/file (file protocol assumes that file already exists on worker host)] from which worker can download the package | True  
+| `classname` | The source's class name if archive is file-url-path (file://) | False
+| `configs` | User defined configs key/values (JSON string) | False
+| `deserialization_classname` | The SerDe classname for the source | False  
+| `processing_guarantees` | Define the message delivery semantics, default to ATLEAST_ONCE (ATLEAST_ONCE, ATMOST_ONCE, EFFECTIVELY_ONCE) | False  
+| `parallelism` | The source's parallelism factor | False  
+| `cpu` | The CPU that needs to be allocated per source instance (applicable only to Docker runtime) | False
+| `ram_mb` | The RAM that need to be allocated per source instance (applicable only to the process and Docker runtimes) | False
+| `disk_mb` | The disk that need to be allocated per source instance (applicable only to Docker runtime) | False
+| `runtime_flags` | User defined configs key/values (JSON string) | False
 
 Importing existing resources
 ------------

bytesize/bytesize.go

@@ -0,0 +1,44 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 bytesize
+
+const MB = 1 << (10 * 2)
+
+type byteSize struct {
+	value uint64
+}
+
+func newByteSize(n uint64) *byteSize {
+	return &byteSize{value: n}
+}
+
+func FormBytes(n uint64) *byteSize {
+	return newByteSize(n)
+}
+
+func FormMegaBytes(n uint64) *byteSize {
+	return newByteSize(n * MB)
+}
+
+func (b *byteSize) ToBytes() uint64 {
+	return b.value
+}
+
+func (b *byteSize) ToMegaBytes() uint64 {
+	return b.value / MB
+}

bytesize/bytesize_test.go

@@ -0,0 +1,52 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 bytesize
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFormBytes(t *testing.T) {
+	testcases := make(map[uint64]uint64)
+	testcases[1] = 1 * 1024 * 1024
+	testcases[2] = 2 * 1024 * 1024
+	testcases[10] = 10 * 1024 * 1024
+	testcases[20] = 20 * 1024 * 1024
+
+	for mb, bytes := range testcases {
+		b := FormBytes(bytes)
+		assert.Equal(t, bytes, b.ToBytes())
+		assert.Equal(t, mb, b.ToMegaBytes())
+	}
+}
+
+func TestFormMegaBytes(t *testing.T) {
+	testcases := make(map[uint64]uint64)
+	testcases[1] = 1 * 1024 * 1024
+	testcases[2] = 2 * 1024 * 1024
+	testcases[10] = 10 * 1024 * 1024
+	testcases[20] = 20 * 1024 * 1024
+
+	for mb, bytes := range testcases {
+		b := FormMegaBytes(mb)
+		assert.Equal(t, bytes, b.ToBytes())
+		assert.Equal(t, mb, b.ToMegaBytes())
+	}
+}

examples/sources/main.tf

@@ -0,0 +1,50 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+terraform {
+  required_providers {
+    pulsar = {
+      version = "1.0.0"
+      source = "registry.terraform.io/apache/pulsar"
+    }
+  }
+}
+
+provider "pulsar" {
+  web_service_url = "http://localhost:8080"
+  api_version = "3"
+}
+
+resource "pulsar_source" "source-1" {
+  provider = pulsar
+
+  name = "source-1"
+  tenant = "public"
+  namespace = "default"
+
+  archive = "https://www.apache.org/dyn/mirrors/mirrors.cgi?action=download&filename=pulsar/pulsar-2.8.1/connectors/pulsar-io-file-2.8.1.nar"
+
+  destination_topic_name = "source-1-topic"
+
+  processing_guarantees = "EFFECTIVELY_ONCE"
+
+  configs = "{\"inputDirectory\":\"opt\"}"
+
+  cpu = 2
+  disk_mb = 20480
+  ram_mb = 2048
+}

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/hashicorp/terraform-plugin-sdk/v2 v2.16.0
 	github.com/pkg/errors v0.9.1
 	github.com/streamnative/pulsarctl v0.4.3-0.20220124120652-ed7f74559899
+	github.com/stretchr/testify v1.7.0
 )
 
 require (
@@ -54,6 +55,7 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/mtibben/percent v0.2.1 // indirect
 	github.com/oklog/run v1.0.0 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/vmihailenco/msgpack v4.0.4+incompatible // indirect
 	github.com/vmihailenco/msgpack/v4 v4.3.12 // indirect
 	github.com/vmihailenco/tagparser v0.1.1 // indirect
@@ -68,4 +70,5 @@ require (
 	google.golang.org/genproto v0.0.0-20200825200019-8632dd797987 // indirect
 	google.golang.org/grpc v1.45.0 // indirect
 	google.golang.org/protobuf v1.28.0 // indirect
+	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
 )

main.go

@@ -19,6 +19,7 @@ package main
 
 import (
 	"github.com/hashicorp/terraform-plugin-sdk/v2/plugin"
+
 	"github.com/streamnative/terraform-provider-pulsar/pulsar"
 )
 

pulsar/provider.go

@@ -76,6 +76,7 @@ func Provider() *schema.Provider {
 			"pulsar_cluster":   resourcePulsarCluster(),
 			"pulsar_namespace": resourcePulsarNamespace(),
 			"pulsar_topic":     resourcePulsarTopic(),
+			"pulsar_source":    resourcePulsarSource(),
 		},
 	}