forked from knative-extensions/eventing-kafka
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[v0.22.0] Backport PR 563 (knative-extensions#233)
* [v0.22.0] Backport of PR 563 (use of knative-extensions#215) * Backport of (knative-sandbox#476) Dispatcher readiness e2e test * Parameterize Partitions created via MustCreateTopic * Add kafkaChannel creation/get obj helper functions * Add subscription scale readiness test * Add explicit t.Run statement * Move test to helper function, use channelTestRunner * Change from subscriptionv1beta1 to subscriptionv1 (abstraction prep) * Support subscription v1beta1 and v1 * Fix copyright header * Change channel creation depnding on typemeta Remove subscription v1beta1 testing, having multiple tests run at once will change the dispatcher replication in parallel * Drop unneeded parameter * Add a blank line * Fix linter complaints * Add build constraint tag to mark testing consolidated channel only * run hack/update-codegen.sh * run hack/update-codegen.sh * Modify build constraints for test * Remove v1alpha1 references as requested * Fix old reference * Change subscription creation function reference * Backport of (knative-sandbox#563) Continual tests for KafkaSource for upgrade testing * Using knative/eventing#4815 to override wathola sender transport * Actual way of replacing an image used for wathola sender * Reformatting * Wathola Kafka Sender utilizing plain connection * Implementing SUT and custom sender * KafkaSource upgrade test * Update knative-eventing to 8270497041f2 * Using source config template * Export kafkaTopicEndpoint to be able to read on TOML * wathola config reader can read both string and map[string]interface{} for sender.address * KafkaSource upgrade test works * Allow to influence Kafka cluster connection details * Channel tests working * Removing unused variable name * Review knative/eventing#5321 changes * Updates after review of knative/eventing#4815 * Use latest integration/eventing-sandbox/eventing-kafka/pr-563 commit * Update deps * Update deps * Refactor after review * Shorten the TypeMeta field * Tweak the merge slices func after review * Adding generated test images Co-authored-by: Lukas Berk <[email protected]> * Restore manually changed vendor files * Skipping subscription readiness when scaling dispatcher test (knative-sandbox#573) Signed-off-by: Ahmed Abdalla <[email protected]> Co-authored-by: Lukas Berk <[email protected]> Co-authored-by: Ahmed Abdalla Abdelrehim <[email protected]>
- Loading branch information
3 people
authored
Jun 10, 2021
1 parent
b6f1e2a
commit 8ce9b2e
Showing
104 changed files
with
7,200 additions
and
5,798 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
5 changes: 5 additions & 0 deletions
5
openshift/ci-operator/knative-test-images/wathola-kafka-sender/Dockerfile
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| # Do not edit! This file was generate via Makefile | ||
| FROM openshift/origin-base | ||
|
|
||
| ADD wathola-kafka-sender /usr/bin/wathola-kafka-sender | ||
| ENTRYPOINT ["/usr/bin/wathola-kafka-sender"] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| // +build e2e | ||
| // +build consolidated | ||
|
|
||
| /* | ||
| Copyright 2021 The Knative Authors | ||
| 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 e2e | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "knative.dev/eventing-kafka/test/e2e/helpers" | ||
| ) | ||
|
|
||
| func TestChannelSubscriptionScaleReadyV1(t *testing.T) { | ||
| t.Skipf("Skipping test due to flakiness.") | ||
| helpers.ChannelSubscriptionScaleReadyHelper(context.Background(), t, channelTestRunner) | ||
| } |
141 changes: 141 additions & 0 deletions
141
test/e2e/helpers/channel_subscription_ready_test_helper.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,141 @@ | ||
| /* | ||
| Copyright 2021 The Knative Authors | ||
| 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 helpers | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "testing" | ||
|
|
||
| cloudevents "github.com/cloudevents/sdk-go/v2" | ||
| . "github.com/cloudevents/sdk-go/v2/test" | ||
| "github.com/google/uuid" | ||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| channelsv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1" | ||
| contribtestlib "knative.dev/eventing-kafka/test/lib" | ||
| testlib "knative.dev/eventing/test/lib" | ||
| "knative.dev/eventing/test/lib/recordevents" | ||
| "knative.dev/eventing/test/lib/resources" | ||
| ) | ||
|
|
||
| const ( | ||
| kafkaChannelName = "kafka-sub-ready-channel" | ||
| kafkaSub0 = "kafka-sub-0" | ||
| kafkaSub1 = "kafka-sub-1" | ||
| recordEventsPodName = "e2e-channel-sub-ready-recordevents-pod" | ||
| eventSenderName = "e2e-channel-event-sender-pod" | ||
| ) | ||
|
|
||
| func scaleDispatcherDeployment(ctx context.Context, t *testing.T, desiredReplicas int32, client *testlib.Client) { | ||
| dispatcherDeployment, err := client.Kube.AppsV1().Deployments("knative-eventing").Get(ctx, "kafka-ch-dispatcher", metav1.GetOptions{}) | ||
| if err != nil { | ||
| t.Fatalf("Unable to get kafka-ch-dispatcher deployment: %v", err) | ||
| } | ||
| if *dispatcherDeployment.Spec.Replicas != desiredReplicas { | ||
| desired := dispatcherDeployment.DeepCopy() | ||
| *desired.Spec.Replicas = desiredReplicas | ||
| _, err := client.Kube.AppsV1().Deployments("knative-eventing").Update(ctx, desired, metav1.UpdateOptions{}) | ||
| if err != nil { | ||
| t.Fatalf("Unable to update kafka-ch-dispatcher deployment to %d replica: %v", desiredReplicas, err) | ||
| } | ||
| // we actively do NOT wait for deployments to be ready so we can check state below | ||
| } | ||
| } | ||
|
|
||
| func readyDispatcherPodsCheck(ctx context.Context, t *testing.T, client *testlib.Client) int32 { | ||
| dispatcherDeployment, err := client.Kube.AppsV1().Deployments("knative-eventing").Get(ctx, "kafka-ch-dispatcher", metav1.GetOptions{}) | ||
| if err != nil { | ||
| t.Fatalf("Unable to get kafka-ch-dispatcher deployment: %v", err) | ||
| } | ||
| return dispatcherDeployment.Status.ReadyReplicas | ||
| } | ||
|
|
||
| func createKafkaChannel(client *testlib.Client, kafkaChannelMeta metav1.TypeMeta, kafkaChannelName string) { | ||
| kafkaChannelV1Beta1 := &channelsv1beta1.KafkaChannel{ | ||
| ObjectMeta: metav1.ObjectMeta{ | ||
| Name: kafkaChannelName, | ||
| }, | ||
| Spec: channelsv1beta1.KafkaChannelSpec{ | ||
| NumPartitions: 3, | ||
| }, | ||
| } | ||
| contribtestlib.CreateKafkaChannelV1Beta1OrFail(client, kafkaChannelV1Beta1) | ||
| } | ||
|
|
||
| func ChannelSubscriptionScaleReadyHelper( | ||
| ctx context.Context, | ||
| t *testing.T, | ||
| channelTestRunner testlib.ComponentsTestRunner, | ||
| options ...testlib.SetupClientOption) { | ||
|
|
||
| eventSource := fmt.Sprintf("http://%s.svc/", eventSenderName) | ||
|
|
||
| channelTestRunner.RunTests(t, testlib.FeatureBasic, func(st *testing.T, kafkaChannelMeta metav1.TypeMeta) { | ||
| client := testlib.Setup(st, true, options...) | ||
| defer testlib.TearDown(client) | ||
|
|
||
| scaleDispatcherDeployment(ctx, st, 1, client) | ||
| createKafkaChannel(client, kafkaChannelMeta, kafkaChannelName) | ||
| client.WaitForResourceReadyOrFail(kafkaChannelName, &kafkaChannelMeta) | ||
|
|
||
| eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, recordEventsPodName) | ||
| client.CreateSubscriptionOrFail( | ||
| kafkaSub0, | ||
| kafkaChannelName, | ||
| &kafkaChannelMeta, | ||
| resources.WithSubscriberForSubscription(recordEventsPodName), | ||
| ) | ||
| client.WaitForAllTestResourcesReadyOrFail(ctx) | ||
|
|
||
| scaleDispatcherDeployment(ctx, st, 4, client) | ||
| client.WaitForResourceReadyOrFail(kafkaSub0, testlib.SubscriptionTypeMeta) //this should still be ready | ||
|
|
||
| client.CreateSubscriptionOrFail( | ||
| kafkaSub1, | ||
| kafkaChannelName, | ||
| &kafkaChannelMeta, | ||
| resources.WithSubscriberForSubscription(recordEventsPodName), | ||
| ) | ||
| for readyDispatcherPodsCheck(ctx, st, client) < 3 { | ||
| subObj, err := client.Eventing.MessagingV1().Subscriptions(client.Namespace).Get(ctx, kafkaSub1, metav1.GetOptions{}) | ||
| if err != nil { | ||
| st.Fatalf("Could not get v1 subscription object %q: %v", subObj.Name, err) | ||
| } | ||
| if subObj.Status.IsReady() { | ||
| st.Fatalf("Subscription: %s, marked ready before dispatcher pods ready", subObj.Name) | ||
| } | ||
| } | ||
| client.WaitForResourceReadyOrFail(kafkaSub1, testlib.SubscriptionTypeMeta) | ||
| // send CloudEvent to the first channel | ||
| event := cloudevents.NewEvent() | ||
| event.SetID("test") | ||
| event.SetSource(eventSource) | ||
| event.SetType(testlib.DefaultEventType) | ||
|
|
||
| body := fmt.Sprintf(`{"msg":"TestSingleEvent %s"}`, uuid.New().String()) | ||
| if err := event.SetData(cloudevents.ApplicationJSON, []byte(body)); err != nil { | ||
| st.Fatalf("Cannot set the payload of the event: %s", err.Error()) | ||
| } | ||
|
|
||
| client.SendEventToAddressable(ctx, eventSenderName, kafkaChannelName, &kafkaChannelMeta, event) | ||
| // verify the logger service receives the event | ||
| eventTracker.AssertAtLeast(1, recordevents.MatchEvent( | ||
| HasSource(eventSource), | ||
| HasData([]byte(body)), | ||
| )) | ||
| }) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| /* | ||
| Copyright 2021 The Knative Authors | ||
| 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 main | ||
|
|
||
| import ( | ||
| "go.uber.org/zap" | ||
| "knative.dev/eventing-kafka/test/upgrade/continual" | ||
| "knative.dev/eventing/test/upgrade/prober/wathola/sender" | ||
| "knative.dev/pkg/signals" | ||
| ) | ||
|
|
||
| func main() { | ||
| ctx := signals.NewContext() | ||
| log, err := zap.NewDevelopment() | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| es := continual.CreateKafkaSender(ctx, log.Sugar()) | ||
| sender.RegisterEventSender(es) | ||
| sender.New().SendContinually() | ||
| } |
Oops, something went wrong.