Skip to content

Commit

Permalink
Consumergroup azure names (#348)
Browse files Browse the repository at this point in the history
* made azure consumer group configurable

* made azure consumer group configurable

* relaxed consumergroup error assertion.

* status code for consumer get

* added error logging if eventhub is missing. reordered suite assertion.

* removed generated code

* renamed to AzureConsumerGroupName
  • Loading branch information
szoio committed Oct 14, 2019
1 parent 6c0f808 commit f74ec73
Show file tree
Hide file tree
Showing 7 changed files with 50 additions and 28 deletions.
7 changes: 4 additions & 3 deletions api/v1alpha1/consumergroup_types.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,10 @@ import (
type ConsumerGroupSpec struct {
// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
// Important: Run "make" to regenerate code after modifying this file
ResourceGroupName string `json:"resourceGroup,omitempty"`
NamespaceName string `json:"namespace,omitempty"`
EventhubName string `json:"eventHub,omitempty"`
ResourceGroupName string `json:"resourceGroup,omitempty"`
NamespaceName string `json:"namespace,omitempty"`
EventhubName string `json:"eventHub,omitempty"`
AzureConsumerGroupName string `json:"consumerGroupName,omitempty"`
}

// ConsumerGroupStatus defines the observed state of ConsumerGroup
Expand Down
9 changes: 8 additions & 1 deletion api/v1alpha1/consumergroup_types_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,14 @@ var _ = Describe("ConsumerGroup", func() {
ObjectMeta: metav1.ObjectMeta{
Name: "foo",
Namespace: "default",
}}
},
Spec: ConsumerGroupSpec{
ResourceGroupName: "rg",
NamespaceName: "ns",
EventhubName: "eh",
AzureConsumerGroupName: "cg",
},
}

By("creating an API obj")
Expect(k8sClient.Create(context.TODO(), created)).To(Succeed())
Expand Down
9 changes: 0 additions & 9 deletions api/v1alpha1/eventhub_types.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,15 +23,6 @@ import (
// EDIT THIS FILE! THIS IS SCAFFOLDING FOR YOU TO OWN!
// NOTE: json tags are required. Any new fields you add must have json tags for the fields to be serialized.

//EventhubNamespaceResource defines the namespace
type EventhubNamespaceResource struct {
Name string `json:"name"`
Location string `json:"location"`
Sku EventhubNamespaceSku `json:"sku,omitempty"`
Properties EventhubNamespaceProperties `json:"properties,omitempty"`
ResourceGroupName string `json:"resourceGroup,omitempty"`
}

// EventhubSpec defines the desired state of Eventhub
type EventhubSpec struct {
// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
Expand Down
9 changes: 5 additions & 4 deletions controllers/consumergroup_controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,7 @@ func (r *ConsumerGroupReconciler) createConsumerGroup(instance *azurev1alpha1.Co
namespaceName := instance.Spec.NamespaceName
resourcegroup := instance.Spec.ResourceGroupName
eventhubName := instance.Spec.EventhubName
azureConsumerGroupName := instance.Spec.AzureConsumerGroupName

// write information back to instance
instance.Status.Provisioning = true
Expand All @@ -119,7 +120,7 @@ func (r *ConsumerGroupReconciler) createConsumerGroup(instance *azurev1alpha1.Co

if err != nil {
//log error and kill it, as the parent might not exist in the cluster. It could have been created elsewhere or through the portal directly
r.Recorder.Event(instance, "Warning", "Failed", "Unable to get owner instance of eventhub")
r.Recorder.Event(instance, "Warning", "Failed", fmt.Sprintf("Unable to get owner eventhub '%s' of consumer group '%s'", eventhubName, consumergroupName))
} else {
//set owner reference for consumer group if it exists
references := []metav1.OwnerReference{
Expand All @@ -139,7 +140,7 @@ func (r *ConsumerGroupReconciler) createConsumerGroup(instance *azurev1alpha1.Co
r.Recorder.Event(instance, "Warning", "Failed", "Unable to update instance")
}

_, err = r.ConsumerGroupManager.CreateConsumerGroup(ctx, resourcegroup, namespaceName, eventhubName, consumergroupName)
_, err = r.ConsumerGroupManager.CreateConsumerGroup(ctx, resourcegroup, namespaceName, eventhubName, azureConsumerGroupName)
if err != nil {

r.Recorder.Event(instance, "Warning", "Failed", "Couldn't create consumer group in azure")
Expand Down Expand Up @@ -170,13 +171,13 @@ func (r *ConsumerGroupReconciler) createConsumerGroup(instance *azurev1alpha1.Co
func (r *ConsumerGroupReconciler) deleteConsumerGroup(instance *azurev1alpha1.ConsumerGroup) error {
ctx := context.Background()

consumergroupName := instance.ObjectMeta.Name
namespaceName := instance.Spec.NamespaceName
resourcegroup := instance.Spec.ResourceGroupName
eventhubName := instance.Spec.EventhubName
azureConsumerGroupName := instance.Spec.AzureConsumerGroupName

var err error
_, err = r.ConsumerGroupManager.DeleteConsumerGroup(ctx, resourcegroup, namespaceName, eventhubName, consumergroupName)
_, err = r.ConsumerGroupManager.DeleteConsumerGroup(ctx, resourcegroup, namespaceName, eventhubName, azureConsumerGroupName)
if err != nil {
r.Recorder.Event(instance, "Warning", "Failed", "Couldn't delete consumer group in azure")
return err
Expand Down
21 changes: 18 additions & 3 deletions controllers/consumergroup_controller_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ package controllers

import (
"context"
"net/http"

azurev1alpha1 "github.com/Azure/azure-service-operator/api/v1alpha1"
"github.com/Azure/azure-service-operator/pkg/helpers"
Expand Down Expand Up @@ -54,6 +55,7 @@ var _ = Describe("ConsumerGroup Controller", func() {
It("should create and delete consumer groups", func() {

consumerGroupName := "t-cg-" + helpers.RandomString(10)
azureConsumerGroupName := consumerGroupName + "-azure"

var err error

Expand All @@ -64,9 +66,10 @@ var _ = Describe("ConsumerGroup Controller", func() {
Namespace: "default",
},
Spec: azurev1alpha1.ConsumerGroupSpec{
NamespaceName: ehnName,
ResourceGroupName: rgName,
EventhubName: ehName,
NamespaceName: ehnName,
ResourceGroupName: rgName,
EventhubName: ehName,
AzureConsumerGroupName: azureConsumerGroupName,
},
}

Expand All @@ -88,6 +91,12 @@ var _ = Describe("ConsumerGroup Controller", func() {
}, tc.timeout,
).Should(BeTrue())

Eventually(func() bool {
cg, _ := tc.eventHubManagers.ConsumerGroup.GetConsumerGroup(context.Background(), rgName, ehnName, ehName, azureConsumerGroupName)
return cg.Name != nil && *cg.Name == azureConsumerGroupName && cg.Response.StatusCode == http.StatusOK
}, tc.timeout,
).Should(BeTrue())

err = tc.k8sClient.Delete(context.Background(), consumerGroupInstance)
Expect(err).NotTo(HaveOccurred())

Expand All @@ -97,6 +106,12 @@ var _ = Describe("ConsumerGroup Controller", func() {
}, tc.timeout,
).Should(BeTrue())

Eventually(func() bool {
cg, _ := tc.eventHubManagers.ConsumerGroup.GetConsumerGroup(context.Background(), rgName, ehnName, ehName, azureConsumerGroupName)
return cg.Response.StatusCode != http.StatusOK
}, tc.timeout,
).Should(BeTrue())

})
})
})
18 changes: 11 additions & 7 deletions controllers/suite_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -215,16 +215,26 @@ var _ = BeforeSuite(func() {
eventHubNSManager := eventHubManagers.EventHubNamespace
// Create the Eventhub namespace resource
_, err = eventHubNSManager.CreateNamespaceAndWait(context.Background(), resourceGroupName, eventhubNamespaceName, namespaceLocation)
Expect(err).ToNot(HaveOccurred())

Eventually(func() bool {
namespace, _ := eventHubManagers.EventHubNamespace.GetNamespace(context.Background(), resourceGroupName, eventhubNamespaceName)
return namespace.ProvisioningState != nil && *namespace.ProvisioningState == "Succeeded"
}, 60,
).Should(BeTrue())

// Create the Eventhub resource
_, err = eventHubManagers.EventHub.CreateHub(context.Background(), resourceGroupName, eventhubNamespaceName, eventhubName, int32(7), int32(1), nil)
_, err = eventHubManagers.EventHub.CreateHub(context.Background(), resourceGroupName, eventhubNamespaceName, eventhubName, int32(7), int32(2), nil)
Expect(err).ToNot(HaveOccurred())

// Create the Storage Account and Container
_, err = storageManagers.Storage.CreateStorage(context.Background(), resourceGroupName, storageAccountName, resourcegroupLocation, azurev1alpha1.StorageSku{
Name: "Standard_LRS",
}, "Storage", map[string]*string{}, "", nil)
Expect(err).ToNot(HaveOccurred())

_, err = storageManagers.BlobContainer.CreateBlobContainer(context.Background(), resourceGroupName, storageAccountName, blobContainerName)
Expect(err).ToNot(HaveOccurred())

tc = testContext{
k8sClient: k8sClient,
Expand All @@ -241,12 +251,6 @@ var _ = BeforeSuite(func() {
keyVaultManager: keyVaultManager,
timeout: timeout,
}

Eventually(func() bool {
namespace, _ := eventHubManagers.EventHubNamespace.GetNamespace(context.Background(), resourceGroupName, eventhubNamespaceName)
return namespace.ProvisioningState != nil && *namespace.ProvisioningState == "Succeeded"
}, 60,
).Should(BeTrue())
})

var _ = AfterSuite(func() {
Expand Down
5 changes: 4 additions & 1 deletion pkg/resourcemanager/mock/eventhubs/consumergroup.go
Original file line number Diff line number Diff line change
Expand Up @@ -102,8 +102,11 @@ func (manager *mockConsumerGroupManager) GetConsumerGroup(ctx context.Context, r
})

if index == -1 {
return eventhub.ConsumerGroup{}, errors.New("consumer group not found")
return eventhub.ConsumerGroup{
Response: helpers.GetRestResponse(http.StatusNotFound),
}, errors.New("consumer group not found")
}

group.ConsumerGroup.Response = helpers.GetRestResponse(http.StatusOK)
return group.ConsumerGroup, nil
}

0 comments on commit f74ec73

Please sign in to comment.