From e01483bff7d58486d4baae856cb8a958d393c808 Mon Sep 17 00:00:00 2001 From: Korbinian Stoemmer Date: Sat, 12 Aug 2023 15:21:35 +0200 Subject: [PATCH] simplify loadtester code (#23) * reconcile subscriptions * simplify loadtest * add subscriptions * synchronize id generation * change base config * code cleanup * add unit test github action * add build action * check if events get sent during test --- internal/loadtest/config/config.go | 62 +-- internal/loadtest/config/notifiable.go | 6 + .../loadtest/events/GenericEvent/events.go | 203 ---------- .../loadtest/events/GenericFactory/factory.go | 58 --- internal/loadtest/events/events.go | 30 -- internal/loadtest/events/factory.go | 145 +++++++ internal/loadtest/events/factory_test.go | 83 ++++ internal/loadtest/events/generator.go | 130 ++++++ internal/loadtest/publisher/publisher.go | 33 +- internal/loadtest/sender/cloudevent/sender.go | 59 +-- .../sender/{ => interface}/interface.go | 11 +- .../loadtest/sender/legacyevent/sender.go | 73 ++-- internal/loadtest/sender/sender.go | 377 ++++++------------ resources/loadtest/base/configmap.yaml | 2 +- resources/loadtest/base/kustomization.yaml | 1 + resources/loadtest/base/subscriptions.yaml | 47 +++ 16 files changed, 644 insertions(+), 676 deletions(-) delete mode 100644 internal/loadtest/events/GenericEvent/events.go delete mode 100644 internal/loadtest/events/GenericFactory/factory.go delete mode 100644 internal/loadtest/events/events.go create mode 100644 internal/loadtest/events/factory.go create mode 100644 internal/loadtest/events/factory_test.go create mode 100644 internal/loadtest/events/generator.go rename internal/loadtest/sender/{ => interface}/interface.go (58%) create mode 100644 resources/loadtest/base/subscriptions.yaml diff --git a/internal/loadtest/config/config.go b/internal/loadtest/config/config.go index 832da24..48db1bf 100644 --- a/internal/loadtest/config/config.go +++ b/internal/loadtest/config/config.go @@ -3,33 +3,17 @@ package config import ( "flag" "fmt" - "strings" "time" ) // Config represents environment config. type Config struct { - ServerAddress string - PublishEndpoint string `config:"publish_endpoint"` - UseLegacyEvents bool `config:"use_legacy_events"` - EventSource string `config:"event_source"` - MaxInflightMessages0 string `config:"max_inflight_messages_0"` - MaxInflightMessages1 string `config:"max_inflight_messages_1"` - EventName0 string `config:"event_name_0"` - EventName1 string `config:"event_name_1"` - VersionFormat string `config:"version_format"` - GenerateCount0 int `config:"generate_count_0"` - GenerateCount1 int `config:"generate_count_1"` - EpsStart0 int `config:"eps_start_0"` - EpsStart1 int `config:"eps_start_1"` - EpsIncrement0 int `config:"eps_increment_0"` - EpsIncrement1 int `config:"eps_increment_1"` - EpsLimit int `config:"eps_limit"` - Workers int `config:"workers"` - MaxIdleConns int `config:"max_idle_conns"` - MaxConnsPerHost int `config:"max_conns_per_host"` - MaxIdleConnsPerHost int `config:"max_idle_conns_per_host"` - IdleConnTimeout time.Duration `config:"idle_conn_timeout"` + ServerAddress string + PublishHost string `config:"publish_host"` + MaxIdleConns int `config:"max_idle_conns"` + MaxConnsPerHost int `config:"max_conns_per_host"` + MaxIdleConnsPerHost int `config:"max_idle_conns_per_host"` + IdleConnTimeout time.Duration `config:"idle_conn_timeout"` } func New() *Config { @@ -39,39 +23,9 @@ func New() *Config { return c } -// IsVersionFormatEmpty returns true if event version is empty. -func (c *Config) IsVersionFormatEmpty() bool { - return len(strings.TrimSpace(c.VersionFormat)) == 0 -} - -// IsEventName0Empty returns true if event name 0 is empty. -func (c *Config) IsEventName0Empty() bool { - return len(strings.TrimSpace(c.EventName0)) == 0 -} - -// IsEventName1Empty returns true if event name 1 is empty. -func (c *Config) IsEventName1Empty() bool { - return len(strings.TrimSpace(c.EventName1)) == 0 -} - -// ComputeEventsCount returns the count of events to generate. -func (c *Config) ComputeEventsCount() int { - count := 0 - if c.IsVersionFormatEmpty() { - return count - } - if !c.IsEventName0Empty() { - count += c.GenerateCount0 - } - if !c.IsEventName1Empty() { - count += c.GenerateCount1 - } - return count -} - func (c *Config) String() string { return fmt.Sprintf( - "ServerAddress: %v PublishEndpoint: %v MaxInflightMessages0: %v MaxInflightMessages1: %v EventFormat0: %v EventFormat1: %v GenerateCount0: %v GenerateCount1: %v EpsStart0: %v EpsStart1: %v EpsIncrement0: %v EpsIncrement1: %v EpsLimit: %v Workers: %v MaxIdleConns: %v MaxConnsPerHost : %v MaxIdleConnsPerHost: %v IdleConnTimeout: %v", - c.ServerAddress, c.PublishEndpoint, c.MaxInflightMessages0, c.MaxInflightMessages1, c.EventName0, c.EventName1, c.GenerateCount0, c.GenerateCount1, c.EpsStart0, c.EpsStart1, c.EpsIncrement0, c.EpsIncrement1, c.EpsLimit, c.Workers, c.MaxIdleConns, c.MaxConnsPerHost, c.MaxIdleConnsPerHost, c.IdleConnTimeout, + "ServerAddress: %v PublishHost: %v MaxIdleConns: %v MaxConnsPerHost : %v MaxIdleConnsPerHost: %v IdleConnTimeout: %v", + c.ServerAddress, c.PublishHost, c.MaxIdleConns, c.MaxConnsPerHost, c.MaxIdleConnsPerHost, c.IdleConnTimeout, ) } diff --git a/internal/loadtest/config/notifiable.go b/internal/loadtest/config/notifiable.go index e4999ad..9f2e2bf 100644 --- a/internal/loadtest/config/notifiable.go +++ b/internal/loadtest/config/notifiable.go @@ -2,6 +2,12 @@ package config import corev1 "k8s.io/api/core/v1" +type Notifiable interface { + AddNotifiable + UpdateNotifiable + DeleteNotifiable +} + type AddNotifiable interface { NotifyAdd(*corev1.ConfigMap) } diff --git a/internal/loadtest/events/GenericEvent/events.go b/internal/loadtest/events/GenericEvent/events.go deleted file mode 100644 index dfb147b..0000000 --- a/internal/loadtest/events/GenericEvent/events.go +++ /dev/null @@ -1,203 +0,0 @@ -package GenericEvent - -import ( - "context" - "fmt" - "log" - "sync" - "time" - - cev2 "github.com/cloudevents/sdk-go/v2" - - "github.com/kyma-project/eventing-tools/internal/loadtest/events/payload" - "github.com/kyma-project/eventing-tools/internal/tree" -) - -type Event struct { - source string - version string - name string - eps int - starttime string - feedback chan int - counter chan int - success chan int - events chan *Event - cancel context.CancelFunc - successes *tree.Node - eventtype string - wg *sync.WaitGroup - running bool - stopper sync.Mutex -} - -func (e *Event) Events() <-chan *Event { - return e.events -} - -func (e *Event) Source() string { - return e.source -} - -func (e *Event) Feedback() chan<- int { - return e.feedback -} - -func (e *Event) Success() chan<- int { - return e.success -} - -func (e *Event) Eps() int { - return e.eps -} - -func (e *Event) Counter() <-chan int { - return e.counter - -} - -func NewEvent(format, name, source string, eps int) *Event { - e := Event{ - version: format, - name: name, - eps: eps, - starttime: time.Now().Format("2006-01-02T15:04:05"), - source: source, - eventtype: fmt.Sprintf("%s.%s", name, format), - wg: &sync.WaitGroup{}, - } - return &e -} - -func (e *Event) handleSuccess(ctx context.Context) { - defer e.wg.Done() - for { - select { - case <-ctx.Done(): - fmt.Printf("%v.%v: %v\n", e.starttime, e.name, e.successes) - fmt.Printf("DONE success %v.%v\n", e.source, e.eventtype) - return - case val := <-e.success: - e.successes = tree.InsertInt(e.successes, val) - } - } -} - -func (e *Event) PrintStats() string { - return fmt.Sprintf("%v.%v.%v.%v: %v\n", e.starttime, e.source, e.name, e.version, e.successes) -} - -func (e *Event) fillCounter(ctx context.Context) { - defer e.wg.Done() - var c int - var next int - for { - select { - case next = <-e.feedback: - break - default: - next = c - c++ - } - select { - case <-ctx.Done(): - fmt.Printf("DONE counter %v.%v\n", e.source, e.eventtype) - return - case e.counter <- next: - break - } - } -} - -func (e *Event) queueEvent(ctx context.Context) { - defer e.wg.Done() - defer func() { - if r := recover(); r != nil { - log.Println("recovered from in queueEvent: ", r) - } - }() - - t := time.NewTicker(time.Second) - defer t.Stop() - - // queue event immediately - for { - select { - case <-t.C: - for i := 0; i < e.eps; i++ { - select { - case <-ctx.Done(): - close(e.events) - fmt.Printf("DONE queue %v.%v\n", e.source, e.eventtype) - return - case e.events <- e: - continue - } - } - case <-ctx.Done(): - close(e.events) - fmt.Printf("DONE queue %v.%v\n", e.source, e.eventtype) - return - } - } -} - -func (e *Event) Stop() { - e.stopper.Lock() - if !e.running { - return - } - e.cancel() - fmt.Printf("waiting for %v.%v\n", e.source, e.eventtype) - e.wg.Wait() - fmt.Printf("DONE waiting for %v.%v\n", e.source, e.eventtype) - e.running = false - e.stopper.Unlock() -} - -func (e *Event) Start() { - if e.running { - return - } - e.running = true - e.events = make(chan *Event, e.eps) - e.counter = make(chan int, e.eps*4) - e.feedback = make(chan int, e.eps*4) - e.success = make(chan int, e.eps*4) - ctx, cancel := context.WithCancel(context.Background()) - e.cancel = cancel - e.successes = nil - e.wg.Add(1) - go e.fillCounter(ctx) - e.wg.Add(1) - go e.handleSuccess(ctx) - e.wg.Add(1) - go e.queueEvent(ctx) -} - -func (e *Event) ToLegacyEvent(seq int) payload.LegacyEvent { - d := payload.DTO{ - Start: e.starttime, - Value: seq, - } - return payload.LegacyEvent{ - Data: d, - EventType: e.name, - EventTypeVersion: e.version, - EventTime: time.Now().Format("2006-01-02T15:04:05.000Z"), - EventTracing: true, - } -} - -func (e *Event) ToCloudEvent(seq int) (cev2.Event, error) { - - ce := cev2.NewEvent() - ce.SetType(e.eventtype) - ce.SetSource(e.source) - d := payload.DTO{ - Start: e.starttime, - Value: seq, - } - err := ce.SetData(cev2.ApplicationJSON, d) - return ce, err -} diff --git a/internal/loadtest/events/GenericFactory/factory.go b/internal/loadtest/events/GenericFactory/factory.go deleted file mode 100644 index a11a17a..0000000 --- a/internal/loadtest/events/GenericFactory/factory.go +++ /dev/null @@ -1,58 +0,0 @@ -package GenericFactory - -import ( - "fmt" - "regexp" - "strconv" - "strings" - - "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" - - "github.com/kyma-project/eventing-tools/internal/loadtest/api/subscription/v1alpha2" - "github.com/kyma-project/eventing-tools/internal/loadtest/events/GenericEvent" -) - -type GenericEventFactory struct { -} - -const ( - formatLabel = "eventing-loadtest" -) - -func New() *GenericEventFactory { - return &GenericEventFactory{} -} - -func (g *GenericEventFactory) FromSubscription(subscription *unstructured.Unstructured, eventFormat string) []*GenericEvent.Event { - sub, err := v1alpha2.ToSubscription(subscription) - if err != nil { - return nil - } - events := []*GenericEvent.Event{} - - // for now we support only type matching standard - if sub.Spec.TypeMatching != v1alpha2.Standard { - return events - } - - if sub.GetLabels()[formatLabel] != eventFormat { - return events - } - - re := regexp.MustCompile(`.v(\d+)$`) - for _, et := range sub.Spec.Types { - if !re.MatchString(et) { - continue - } - rss := re.FindStringSubmatch(et) - for _, rs := range rss { - rate, err := strconv.Atoi(rs) - if err != nil { - continue - } - name := strings.TrimSuffix(et, fmt.Sprintf(".v%v", rs)) // trim the `.v` to get a clean event type without trailing `.`. - events = append(events, GenericEvent.NewEvent(fmt.Sprintf("v%v", rs), name, sub.Spec.Source, rate)) // create a clean `v` version indicator. - } - } - return events -} diff --git a/internal/loadtest/events/events.go b/internal/loadtest/events/events.go deleted file mode 100644 index 5b960b8..0000000 --- a/internal/loadtest/events/events.go +++ /dev/null @@ -1,30 +0,0 @@ -package events - -import ( - cev2 "github.com/cloudevents/sdk-go/v2" - "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" - - "github.com/kyma-project/eventing-tools/internal/loadtest/events/GenericEvent" - "github.com/kyma-project/eventing-tools/internal/loadtest/events/payload" -) - -const ( - LegacyFormat = "legacy" - CloudeventFormat = "cloudevent" -) - -type Event interface { - ToCloudEvent(int, string) (cev2.Event, error) - ToLegacyEvent(int) payload.LegacyEvent - Stop() - Eps() int - Counter() <-chan int - Source() string - Feedback() chan<- int - Success() chan<- int - Events() <-chan Event -} - -type EventFactory interface { - FromSubscription(*unstructured.Unstructured, string) []*GenericEvent.Event -} diff --git a/internal/loadtest/events/factory.go b/internal/loadtest/events/factory.go new file mode 100644 index 0000000..846c15e --- /dev/null +++ b/internal/loadtest/events/factory.go @@ -0,0 +1,145 @@ +package events + +import ( + "regexp" + "strconv" + + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + + "github.com/kyma-project/eventing-tools/internal/loadtest/api/subscription/v1alpha2" + "github.com/kyma-project/eventing-tools/internal/loadtest/subscription" +) + +type eventGenerator map[string]*Generator + +type Factory struct { + generators map[NamespaceName]eventGenerator + senderC chan<- Event +} + +func (f *Factory) OnNewSubscription(subscription *unstructured.Unstructured) { + f.StartReconcile(subscription) +} + +func (f *Factory) OnChangedSubscription(subscription *unstructured.Unstructured) { + f.StartReconcile(subscription) +} + +func (f *Factory) OnDeleteSubscription(subscription *unstructured.Unstructured) { + f.StartReconcile(subscription) +} + +func NewGeneratorFactory(senderC chan<- Event) *Factory { + return &Factory{ + generators: map[NamespaceName]eventGenerator{}, + senderC: senderC, + } +} + +type NamespaceName struct { + Name, Namespace string +} + +const ( + formatLabel = "eventing-loadtest" +) + +var _ subscription.Notifiable = &Factory{} + +func (f *Factory) StartReconcile(subscription *unstructured.Unstructured) { + sub, err := v1alpha2.ToSubscription(subscription) + if err != nil { + // // TODO[k15r]: log + return + } + //nolint:errcheck // TODO[k15r]: get rid of these errors + f.reconcile(sub) +} + +func (f *Factory) reconcile(sub *v1alpha2.Subscription) error { + // for now we support only type matching standard + if sub.Spec.TypeMatching != v1alpha2.Standard { + return nil + } + nn := NamespaceName{ + Name: sub.Name, + Namespace: sub.Namespace, + } + + // delete subscription + if sub.DeletionTimestamp != nil { + if g, ok := f.generators[nn]; ok { + f.stopGenerators(g) + delete(f.generators, nn) + return nil + } + } + + // create default eventGenerator (empty) + if _, ok := f.generators[nn]; !ok { + f.generators[nn] = eventGenerator{} + } + + eg := f.generators[nn] + + // check if eventtypes have been removed +currentGenerator: + for etgen, gen := range eg { + for _, et := range sub.Spec.Types { + if et == etgen { + continue currentGenerator + } + } + // remove generators for removed eventType + gen.Stop() + delete(eg, etgen) + } + + // handle adding EventTypes +currentEventType: + for _, et := range sub.Spec.Types { + //for etgen, gen := range eg { + for etgen, gen := range eg { + if et == etgen { + // update the eventFormat to the one specified in the subscription + updateGeneratorFormat(sub, gen) + continue currentEventType + } + } + // create a new EventGenerator for the found eventType and start it + gen := f.ConfigureAndStartGenerator(sub, et) + if gen != nil { + eg[et] = gen + } + } + return nil +} + +func (f *Factory) Stop() { + for _, gens := range f.generators { + for _, gen := range gens { + gen.Stop() + } + } +} + +func (f *Factory) ConfigureAndStartGenerator(sub *v1alpha2.Subscription, eventType string) *Generator { + re := regexp.MustCompile(`.v(\d+)$`) + if !re.MatchString(eventType) { + return nil + } + rs := re.FindStringSubmatch(eventType)[1] + rate, err := strconv.Atoi(rs) + if err != nil { + return nil + } + gen := NewGenerator(eventType, sub.Spec.Source, rate, sub.GetLabels()[formatLabel], f.senderC) // create a clean `v` version indicator. + gen.Start() + return gen +} + +func (f *Factory) stopGenerators(generators eventGenerator) { + for _, g := range generators { + g.Stop() + } +} diff --git a/internal/loadtest/events/factory_test.go b/internal/loadtest/events/factory_test.go new file mode 100644 index 0000000..c8c7f0a --- /dev/null +++ b/internal/loadtest/events/factory_test.go @@ -0,0 +1,83 @@ +package events + +import ( + "testing" + "time" + + "golang.org/x/net/context" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/kyma-project/eventing-tools/internal/loadtest/api/subscription/v1alpha2" +) + +func TestFactory_reconcile(t *testing.T) { + type fields struct { + generators map[NamespaceName]eventGenerator + senderC chan Event + } + type args struct { + sub *v1alpha2.Subscription + } + tests := []struct { + name string + fields fields + args args + wantErr bool + checkEventsFunc func([]Event) bool + }{ + {name: "new subscription - starts sending events", + fields: fields{ + generators: map[NamespaceName]eventGenerator{}, + senderC: make(chan Event), + }, + args: args{ + sub: &v1alpha2.Subscription{ + ObjectMeta: metav1.ObjectMeta{ + Name: "new", + Namespace: "new", + }, + Spec: v1alpha2.SubscriptionSpec{ + Sink: "", + TypeMatching: v1alpha2.Standard, + Source: "Source", + Types: []string{"foo.bar.v1", "bar.foo.v1"}, + Config: nil, + }, + }, + }, + wantErr: false, + checkEventsFunc: func(events []Event) bool { + return len(events) > 0 + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + var receivedEvents []Event + go func(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case e := <-tt.fields.senderC: + receivedEvents = append(receivedEvents, e) + } + } + }(ctx) + f := &Factory{ + generators: tt.fields.generators, + senderC: tt.fields.senderC, + } + if err := f.reconcile(tt.args.sub); (err != nil) != tt.wantErr { + t.Errorf("reconcile() error = %v, wantErr %v", err, tt.wantErr) + } + time.Sleep(1 * time.Second) + f.Stop() + cancel() + if !tt.checkEventsFunc(receivedEvents) { + t.Fail() + } + }) + } +} diff --git a/internal/loadtest/events/generator.go b/internal/loadtest/events/generator.go new file mode 100644 index 0000000..0edc4a3 --- /dev/null +++ b/internal/loadtest/events/generator.go @@ -0,0 +1,130 @@ +package events + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/kyma-project/eventing-tools/internal/loadtest/api/subscription/v1alpha2" +) + +type Generator struct { + source string + eps int + starttime string + cancel context.CancelFunc + eventtype string + id int + sink string + c chan<- Event + wg sync.WaitGroup + format EventFormat + lock sync.Mutex +} + +type Event struct { + EventType string + Source string + Sink string + ID int + StartTime string + Format EventFormat +} + +func NewGenerator(eventType, source string, eps int, format string, senderC chan<- Event) *Generator { + e := Generator{ + eps: eps, + starttime: time.Now().Format("2006-01-02T15:04:05"), + source: source, + eventtype: eventType, + format: EventFormatFromString(format), + c: senderC, + } + return &e +} + +func updateGeneratorFormat(sub *v1alpha2.Subscription, gen *Generator) { + f := EventFormatFromString(sub.GetLabels()[formatLabel]) + if gen.format != f { + gen.Update(f) + } +} + +func (e *Generator) fillChan(ctx context.Context, c chan<- Event) { + t := time.NewTicker(time.Second) + defer t.Stop() + remaining := e.eps + for { + select { + case <-ctx.Done(): + fmt.Printf("DONE counter %v.%v\n", e.source, e.eventtype) + return + case <-t.C: + remaining = e.eps + default: + if remaining > 0 { + // ensure nowone resets the id atm + e.lock.Lock() + id := e.id + e.id++ + e.lock.Unlock() + + c <- Event{ + EventType: e.eventtype, + Source: e.source, + Sink: e.sink, + ID: id, + StartTime: e.starttime, + Format: e.format, + } + remaining-- + } + } + } +} + +func (e *Generator) Update(format EventFormat) { + // let's stop all concurrency for a while + e.lock.Lock() + e.format = format + e.id = 0 + e.starttime = time.Now().Format("2006-01-02T15:04:05") + e.lock.Unlock() +} + +func (e *Generator) Start() { + ctx, cancel := context.WithCancel(context.Background()) + e.cancel = cancel + e.wg.Add(1) + go func() { + defer e.wg.Done() + e.fillChan(ctx, e.c) + }() +} + +func (e *Generator) Stop() { + e.cancel() + e.wg.Wait() +} + +func EventFormatFromString(format string) EventFormat { + if format == "legacy" { + return Legacy + } + return CloudEvent +} + +type EventFormat int + +func (e EventFormat) String() string { + if e == Legacy { + return "legacy" + } + return "cloudevent" +} + +const ( + Legacy EventFormat = iota + CloudEvent +) diff --git a/internal/loadtest/publisher/publisher.go b/internal/loadtest/publisher/publisher.go index 1b3a704..946aca1 100644 --- a/internal/loadtest/publisher/publisher.go +++ b/internal/loadtest/publisher/publisher.go @@ -8,9 +8,8 @@ import ( "github.com/kyma-project/eventing-tools/internal/k8s" "github.com/kyma-project/eventing-tools/internal/loadtest/config" + "github.com/kyma-project/eventing-tools/internal/loadtest/events" sender2 "github.com/kyma-project/eventing-tools/internal/loadtest/sender" - "github.com/kyma-project/eventing-tools/internal/loadtest/sender/cloudevent" - "github.com/kyma-project/eventing-tools/internal/loadtest/sender/legacyevent" "github.com/kyma-project/eventing-tools/internal/loadtest/subscription" "github.com/kyma-project/eventing-tools/internal/logger" "github.com/kyma-project/eventing-tools/internal/probes" @@ -23,34 +22,26 @@ const ( ) func Start(port int) { - appConfig := config.New() + //appConfig := config.New() k8sConfig := k8s.ConfigOrDie() k8sClient := k8s.ClientOrDie(k8sConfig) dynamicClient := dynamic.NewForConfigOrDie(k8sConfig) - legacySender := legacyevent.NewSender(appConfig) - legacyEventSender := sender2.NewSender(appConfig, legacySender) - - ceSender := cloudevent.NewSender(appConfig) - ceEventSender := sender2.NewSender(appConfig, ceSender) + sender, senderC := sender2.NewSender() + factory := events.NewGeneratorFactory(senderC) config.NewWatcher(k8sClient, Namespace, ConfigMapName). - OnAddNotify(legacyEventSender). - OnUpdateNotify(legacyEventSender). - OnDeleteNotify(legacyEventSender). - OnAddNotify(ceEventSender). - OnUpdateNotify(ceEventSender). - OnDeleteNotify(ceEventSender). - OnDeleteNotifyMe(). + OnAddNotify(sender). + OnUpdateNotify(sender). + OnDeleteNotify(sender). Watch() + sender.Start() + subscription.NewWatcher(dynamicClient). - OnAddNotify(ceEventSender). - OnUpdateNotify(ceEventSender). - OnDeleteNotify(ceEventSender). - OnAddNotify(legacyEventSender). - OnUpdateNotify(legacyEventSender). - OnDeleteNotify(legacyEventSender). + OnAddNotify(factory). + OnUpdateNotify(factory). + OnDeleteNotify(factory). Watch() http.HandleFunc(probes.EndpointReadyz, probes.DefaultHandler) diff --git a/internal/loadtest/sender/cloudevent/sender.go b/internal/loadtest/sender/cloudevent/sender.go index dea14b2..2334ff4 100644 --- a/internal/loadtest/sender/cloudevent/sender.go +++ b/internal/loadtest/sender/cloudevent/sender.go @@ -3,68 +3,77 @@ package cloudevent import ( "context" "fmt" - "net/http" cev2 "github.com/cloudevents/sdk-go/v2" "github.com/cloudevents/sdk-go/v2/client" "github.com/kyma-project/eventing-tools/internal/client/cloudevents" + "github.com/kyma-project/eventing-tools/internal/client/transport" "github.com/kyma-project/eventing-tools/internal/loadtest/config" "github.com/kyma-project/eventing-tools/internal/loadtest/events" - "github.com/kyma-project/eventing-tools/internal/loadtest/events/GenericEvent" - "github.com/kyma-project/eventing-tools/internal/loadtest/sender" + "github.com/kyma-project/eventing-tools/internal/loadtest/events/payload" + "github.com/kyma-project/eventing-tools/internal/loadtest/sender/interface" ) // compile-time check for interfaces implementation. -var _ sender.Sender = &Sender{} +var _ _interface.Sender = &Sender{} // Sender sends cloud events. type Sender struct { - client client.Client - config *config.Config + ackC, nackC, undeliveredC chan<- events.Event + client client.Client + config config.Config } -func (s *Sender) Format() string { - return events.CloudeventFormat +func (s *Sender) Format() events.EventFormat { + return events.CloudEvent } -func (s *Sender) Init(t *http.Transport, cfg *config.Config) { - s.config = cfg +func NewSender(conf config.Config, ackC, nackC, undeliveredC chan<- events.Event) *Sender { + t := transport.New(conf.MaxIdleConns, conf.MaxConnsPerHost, conf.MaxIdleConnsPerHost, conf.IdleConnTimeout) + s := &Sender{ + config: conf, + ackC: ackC, + nackC: nackC, + undeliveredC: undeliveredC, + } s.client = cloudevents.NewClientOrDie(t) -} - -func NewSender(conf *config.Config) *Sender { - s := &Sender{config: conf} return s } -func (s *Sender) SendEvent(evt *GenericEvent.Event, ack chan<- int, nack chan<- int, undelivered chan<- int) { - - seq := <-evt.Counter() +func ToCloudEvent(event events.Event) (cev2.Event, error) { + ce := cev2.NewEvent() + ce.SetType(event.EventType) + ce.SetSource(event.Source) + d := payload.DTO{ + Start: event.StartTime, + Value: event.ID, + } + err := ce.SetData(cev2.ApplicationJSON, d) + return ce, err +} - ce, err := evt.ToCloudEvent(seq) +func (s *Sender) SendEvent(event events.Event) { + ce, err := ToCloudEvent(event) if err != nil { return } - endpoint := fmt.Sprintf("%v/publish", s.config.PublishEndpoint) + endpoint := fmt.Sprintf("%v/publish", s.config.PublishHost) ctx := cev2.WithEncodingStructured(cev2.ContextWithTarget(context.Background(), endpoint)) resp := s.client.Send(ctx, ce) switch { case cev2.IsUndelivered(resp): { - undelivered <- 1 - evt.Feedback() <- seq + s.undeliveredC <- event } case cev2.IsACK(resp): { - ack <- 1 - evt.Success() <- seq + s.ackC <- event } case cev2.IsNACK(resp): { - nack <- 1 - evt.Feedback() <- seq + s.nackC <- event } } } diff --git a/internal/loadtest/sender/interface.go b/internal/loadtest/sender/interface/interface.go similarity index 58% rename from internal/loadtest/sender/interface.go rename to internal/loadtest/sender/interface/interface.go index 1247d1c..d9fb537 100644 --- a/internal/loadtest/sender/interface.go +++ b/internal/loadtest/sender/interface/interface.go @@ -1,10 +1,8 @@ -package sender +package _interface import ( - "net/http" - "github.com/kyma-project/eventing-tools/internal/loadtest/config" - "github.com/kyma-project/eventing-tools/internal/loadtest/events/GenericEvent" + "github.com/kyma-project/eventing-tools/internal/loadtest/events" ) type ConfigHandler interface { @@ -14,7 +12,6 @@ type ConfigHandler interface { } type Sender interface { - SendEvent(e *GenericEvent.Event, ack chan<- int, nack chan<- int, undelivered chan<- int) - Format() string - Init(t *http.Transport, cfg *config.Config) + SendEvent(event events.Event) + Format() events.EventFormat } diff --git a/internal/loadtest/sender/legacyevent/sender.go b/internal/loadtest/sender/legacyevent/sender.go index 53188cb..3056057 100644 --- a/internal/loadtest/sender/legacyevent/sender.go +++ b/internal/loadtest/sender/legacyevent/sender.go @@ -7,51 +7,73 @@ import ( "fmt" "io" "net/http" + "strings" + "time" + "github.com/kyma-project/eventing-tools/internal/client/transport" "github.com/kyma-project/eventing-tools/internal/loadtest/config" "github.com/kyma-project/eventing-tools/internal/loadtest/events" - "github.com/kyma-project/eventing-tools/internal/loadtest/events/GenericEvent" - "github.com/kyma-project/eventing-tools/internal/loadtest/sender" + "github.com/kyma-project/eventing-tools/internal/loadtest/events/payload" + "github.com/kyma-project/eventing-tools/internal/loadtest/sender/interface" ) -var _ sender.Sender = &Sender{} +var _ _interface.Sender = &Sender{} // Sender sends legacy events. type Sender struct { - client *http.Client - config *config.Config + ackC, nackC, undeliveredC chan<- events.Event + client *http.Client + config config.Config } -func (s *Sender) Init(t *http.Transport, cfg *config.Config) { - s.config = cfg - s.client = &http.Client{ - Transport: t, - } -} - -func (s *Sender) Format() string { - return events.LegacyFormat +func (s *Sender) Format() events.EventFormat { + return events.Legacy } -func NewSender(conf *config.Config) *Sender { - s := &Sender{config: conf} +func NewSender(cfg config.Config, ackC, nackC, undeliveredC chan<- events.Event) *Sender { + s := &Sender{ + config: cfg, + client: &http.Client{ + Transport: transport.New(cfg.MaxIdleConns, cfg.MaxConnsPerHost, cfg.MaxIdleConnsPerHost, cfg.IdleConnTimeout), + }, + ackC: ackC, + nackC: nackC, + undeliveredC: undeliveredC, + } return s } -func (s *Sender) SendEvent(evt *GenericEvent.Event, ack, nack, undelivered chan<- int) { +func ToLegacyEvent(event events.Event) payload.LegacyEvent { + d := payload.DTO{ + Start: event.StartTime, + Value: event.ID, + } + eventtype, version := splitEventType(event.EventType) + return payload.LegacyEvent{ + Data: d, + EventType: eventtype, + EventTypeVersion: version, + EventTime: time.Now().Format("2006-01-02T15:04:05.000Z"), + EventTracing: true, + } +} - seq := <-evt.Counter() +func splitEventType(eventType string) (string, string) { + i := strings.LastIndex(eventType, ".") + return eventType[0:i], eventType[i+1:] +} +func (s *Sender) SendEvent(event events.Event) { // Build a http request out of the legacy event. - le := evt.ToLegacyEvent(seq) + le := ToLegacyEvent(event) b, err := json.Marshal(le) if err != nil { return } r := bytes.NewReader(b) - legacyEndpoint := fmt.Sprintf("%s/%s/v1/events", s.config.PublishEndpoint, evt.Source()) + legacyEndpoint := fmt.Sprintf("%s/%s/v1/events", s.config.PublishHost, event.Source) rq, err := http.NewRequestWithContext(context.TODO(), http.MethodPost, legacyEndpoint, r) if err != nil { return @@ -60,13 +82,12 @@ func (s *Sender) SendEvent(evt *GenericEvent.Event, ack, nack, undelivered chan< // Send the http request. if s.client == nil { - undelivered <- 1 + s.undeliveredC <- event return } resp, err := s.client.Do(rq) if err != nil { - undelivered <- 1 - evt.Feedback() <- seq + s.undeliveredC <- event return } @@ -77,13 +98,11 @@ func (s *Sender) SendEvent(evt *GenericEvent.Event, ack, nack, undelivered chan< switch { case resp.StatusCode/100 == 2: { - ack <- 1 - evt.Success() <- seq + s.ackC <- event } default: { - nack <- 1 - evt.Feedback() <- seq + s.nackC <- event } } } diff --git a/internal/loadtest/sender/sender.go b/internal/loadtest/sender/sender.go index 0712bb5..5bdd1d9 100644 --- a/internal/loadtest/sender/sender.go +++ b/internal/loadtest/sender/sender.go @@ -3,308 +3,185 @@ package sender import ( "context" "fmt" - "log" - "reflect" - "strings" + "os" + "sort" "sync" - "sync/atomic" + "text/tabwriter" "time" corev1 "k8s.io/api/core/v1" - "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" - "github.com/kyma-project/eventing-tools/internal/client/transport" "github.com/kyma-project/eventing-tools/internal/loadtest/config" "github.com/kyma-project/eventing-tools/internal/loadtest/events" - "github.com/kyma-project/eventing-tools/internal/loadtest/events/GenericEvent" - "github.com/kyma-project/eventing-tools/internal/loadtest/events/GenericFactory" - "github.com/kyma-project/eventing-tools/internal/loadtest/subscription" + "github.com/kyma-project/eventing-tools/internal/loadtest/sender/cloudevent" + "github.com/kyma-project/eventing-tools/internal/loadtest/sender/interface" + "github.com/kyma-project/eventing-tools/internal/loadtest/sender/legacyevent" ) // compile-time check for interfaces implementation. -var _ subscription.Notifiable = &EventSender{} +var _ config.Notifiable = &EventSender{} -// Sender sends cloud events. +// Sender sends cloud factories. type EventSender struct { + cfg config.Config ctx context.Context cancel context.CancelFunc - config *config.Config - events map[string][]*GenericEvent.Event - factory events.EventFactory - process chan bool - running bool - undelivered int32 - ack int32 - nack int32 - mapLock sync.RWMutex + ackC, nackC, undeliveredC chan events.Event wg sync.WaitGroup - stopper sync.Mutex - sender Sender - acks, nacks, undelivereds chan int + events chan events.Event + senders []_interface.Sender + limitC chan any + writer *tabwriter.Writer } -func (s *EventSender) FormatName() string { - // TODO implement me - panic("implement me") -} - -func NewSender(conf *config.Config, sender Sender) *EventSender { - s := &EventSender{config: conf} - s.undelivered = 0 - s.ack = 0 - s.nack = 0 - s.events = make(map[string][]*GenericEvent.Event) - s.factory = GenericFactory.New() - s.sender = sender - s.acks = make(chan int) - s.nacks = make(chan int) - s.undelivereds = make(chan int) - - return s -} - -func (s *EventSender) NotifyAdd(cm *corev1.ConfigMap) { - s.stopper.Lock() - defer s.stopper.Unlock() - s.stop() - config.Map(cm, s.config) - t := transport.New(s.config.MaxIdleConns, s.config.MaxConnsPerHost, s.config.MaxIdleConnsPerHost, s.config.IdleConnTimeout) - s.sender.Init(t, s.config) - s.ctx, s.cancel = context.WithCancel(context.TODO()) - s.process = make(chan bool, s.config.EpsLimit) - s.start() -} - -func (s *EventSender) NotifyUpdate(cm *corev1.ConfigMap) { - s.stopper.Lock() - defer s.stopper.Unlock() - s.stop() - config.Map(cm, s.config) - t := transport.New(s.config.MaxIdleConns, s.config.MaxConnsPerHost, s.config.MaxIdleConnsPerHost, s.config.IdleConnTimeout) - s.sender.Init(t, s.config) - s.ctx, s.cancel = context.WithCancel(context.TODO()) - s.process = make(chan bool, s.config.EpsLimit) - s.start() -} +func (s *EventSender) NotifyAdd(configMap *corev1.ConfigMap) { + // TODO update config + config.Map(configMap, &s.cfg) + s.senders = append(s.senders, cloudevent.NewSender(s.cfg, s.ackC, s.nackC, s.undeliveredC)) + s.senders = append(s.senders, legacyevent.NewSender(s.cfg, s.ackC, s.nackC, s.undeliveredC)) -func (s *EventSender) NotifyDelete(*corev1.ConfigMap) { - s.stopper.Lock() - defer s.stopper.Unlock() - s.stop() } -func (s *EventSender) OnNewSubscription(subscription *unstructured.Unstructured) { - ne := s.factory.FromSubscription(subscription, s.sender.Format()) - if len(ne) == 0 { - return - } - s.stopper.Lock() - defer s.stopper.Unlock() - - // s.queue = make(chan events.Event, buffer) - for _, e := range ne { - e.Start() - } - s.mapLock.Lock() - defer s.mapLock.Unlock() - s.events[fmt.Sprintf("%v/%v", subscription.GetNamespace(), subscription.GetName())] = ne +func (s *EventSender) NotifyUpdate(configMap *corev1.ConfigMap) { + //TODO implement me + config.Map(configMap, &s.cfg) + s.senders = make([]_interface.Sender, 0) + s.senders = append(s.senders, cloudevent.NewSender(s.cfg, s.ackC, s.nackC, s.undeliveredC)) + s.senders = append(s.senders, legacyevent.NewSender(s.cfg, s.ackC, s.nackC, s.undeliveredC)) } -func (s *EventSender) OnChangedSubscription(subscription *unstructured.Unstructured) { - if subscription.GetDeletionTimestamp() != nil { - return - } - s.stopper.Lock() - defer s.stopper.Unlock() - for _, e := range s.events[fmt.Sprintf("%v/%v", subscription.GetNamespace(), subscription.GetName())] { - e.Stop() - } - s.mapLock.Lock() - defer s.mapLock.Unlock() - delete(s.events, fmt.Sprintf("%v/%v", subscription.GetNamespace(), subscription.GetName())) - - ne := s.factory.FromSubscription(subscription, s.sender.Format()) - if len(ne) == 0 { - return - } - - for _, e := range ne { - e.Start() - } - s.events[fmt.Sprintf("%v/%v", subscription.GetNamespace(), subscription.GetName())] = ne +func (s *EventSender) NotifyDelete(_ *corev1.ConfigMap) { + s.senders = make([]_interface.Sender, 0) } -func (s *EventSender) OnDeleteSubscription(sub *unstructured.Unstructured) { - s.stopper.Lock() - s.mapLock.Lock() - defer s.mapLock.Unlock() - defer s.stopper.Unlock() - for _, e := range s.events[fmt.Sprintf("%v/%v", sub.GetNamespace(), sub.GetName())] { - e.Stop() +func NewSender() (*EventSender, chan<- events.Event) { + eventsC := make(chan events.Event) + ackC := make(chan events.Event) + nackC := make(chan events.Event) + undeliveredC := make(chan events.Event) + s := &EventSender{ + writer: new(tabwriter.Writer), + events: eventsC, + ackC: ackC, + nackC: nackC, + undeliveredC: undeliveredC, } - delete(s.events, fmt.Sprintf("%v/%v", sub.GetNamespace(), sub.GetName())) -} - -func (s *EventSender) init() { + s.writer.Init(os.Stdout, 8, 8, 0, '\t', tabwriter.AlignRight) + return s, eventsC } -func (s *EventSender) start() { - s.ctx, s.cancel = context.WithCancel(context.TODO()) - s.mapLock.RLock() - for _, subs := range s.events { - for _, e := range subs { - e.Start() - } - } - s.mapLock.RUnlock() - s.sendEventsAsync() +func (s *EventSender) Start() { + s.ctx, s.cancel = context.WithCancel(context.Background()) + s.limitC = make(chan any, 3000) s.wg.Add(1) - go s.refillMaxEps(time.Second) + go func() { + defer s.wg.Done() + s.doAccounting() + }() s.wg.Add(1) - go s.reportUsageAsync(time.Second, 20*time.Second) -} - -func (s *EventSender) stop() { - // recover from closing already closed channels - defer func() { - if r := recover(); r != nil { - log.Println("recovered from: ", r) - } + go func() { + defer s.wg.Done() + s.sendEvents() }() - s.mapLock.RLock() - for _, subs := range s.events { - for _, e := range subs { - e.Stop() - } - } - s.mapLock.RUnlock() +} - s.running = false +func (s *EventSender) Stop() { s.cancel() - close(s.process) s.wg.Wait() } -func (s *EventSender) sendEventsAsync() { - for i := 0; i < s.config.Workers; i++ { - s.wg.Add(1) - go s.sendEvents() +func (s *EventSender) sendEvents() { + for { + select { + case e := <-s.events: + // here we have to actually send messages to the sink + for _, es := range s.senders { + if es.Format() == e.Format { + s.limitC <- struct{}{} + go func() { + es.SendEvent(e) + <-s.limitC + }() + break + } + } + case <-s.ctx.Done(): + return + } } } -func (s *EventSender) reportUsageAsync(send, success time.Duration) { +type stat struct { + acks, nacks, undelivered int +} - defer func() { - s.wg.Done() - }() +type stats map[string]stat - sendt := time.NewTicker(send) - defer sendt.Stop() - succt := time.NewTicker(success) - defer succt.Stop() +func (s *EventSender) doAccounting() { + dur := 10 * time.Second + tickOne := time.NewTicker(1 * time.Second) + tickTen := time.NewTicker(dur) + defer tickOne.Stop() + defer tickTen.Stop() + cs := make(stats) + var all stat + fmt.Fprintf(s.writer, "%s\t%s\t%s", "ACK", "NACK", "UNDELIVERED") + fmt.Fprintf(s.writer, "\n%s\t%s\t%s", "----", "----", "----") + s.writer.Flush() for { select { - case na := <-s.acks: - atomic.AddInt32(&s.ack, int32(na)) - case nn := <-s.nacks: - atomic.AddInt32(&s.nack, int32(nn)) - case nu := <-s.undelivereds: - atomic.AddInt32(&s.undelivered, int32(nu)) - + case <-tickOne.C: + fmt.Fprintf(s.writer, "\n%v\t%v\t%v", all.acks, all.nacks, all.undelivered) + s.writer.Flush() + all = stat{} + case <-tickTen.C: + s.printStats(cs, dur) + cs = make(stats) + case e := <-s.ackC: + all.acks++ + st := cs["["+e.Format.String()+"]"+e.Source+"/"+e.EventType] + st.acks++ + cs["["+e.Format.String()+"]"+e.Source+"/"+e.EventType] = st + case e := <-s.nackC: + all.nacks++ + st := cs["["+e.Format.String()+"]"+e.Source+"/"+e.EventType] + st.nacks++ + cs["["+e.Format.String()+"]"+e.Source+"/"+e.EventType] = st + case e := <-s.undeliveredC: + all.undelivered++ + st := cs["["+e.Format.String()+"]"+e.Source+"/"+e.EventType] + st.undelivered++ + cs["["+e.Format.String()+"]"+e.Source+"/"+e.EventType] = st case <-s.ctx.Done(): - targetEPS := s.ComputeTotalEventsPerSecond() - log.Printf( - "%v: | target_eps:% 4d (% 4d)| undelivered:% 4d | ack:% 4d | nack:% 4d | sum:% 4d |", - s.sender.Format(), targetEPS, s.config.EpsLimit, s.undelivered, s.ack, s.nack, s.undelivered+s.ack+s.nack, - ) + fmt.Fprintf(s.writer, "\n%v\t%v\t%v\n", all.acks, all.nacks, all.undelivered) + s.writer.Flush() + s.printStats(cs, dur) return - case <-sendt.C: - targetEPS := s.ComputeTotalEventsPerSecond() - if targetEPS == 0 { - continue - } - log.Printf( - "%v: | target_eps:% 4d (% 4d)| undelivered:% 4d | ack:% 4d | nack:% 4d | sum:% 4d |", - s.sender.Format(), targetEPS, s.config.EpsLimit, s.undelivered, s.ack, s.nack, s.undelivered+s.ack+s.nack, - ) - // reset counts for last report - atomic.StoreInt32(&s.undelivered, 0) - atomic.StoreInt32(&s.ack, 0) - atomic.StoreInt32(&s.nack, 0) - case <-succt.C: - s.mapLock.RLock() - stats := []string{fmt.Sprintf("%v:", s.sender.Format())} - for _, subs := range s.events { - for _, e := range subs { - stats = append(stats, e.PrintStats()) - } - } - fmt.Println(strings.Join(stats, "\n\t")) - s.mapLock.RUnlock() } } } -func (s *EventSender) sendEvents() { - for { - var cases []reflect.SelectCase - s.mapLock.RLock() - for _, subs := range s.events { - for _, e := range subs { - if e.Events() != nil { - cases = append(cases, reflect.SelectCase{ - Dir: reflect.SelectRecv, - Chan: reflect.ValueOf(e.Events()), - }) - } - } - } - s.mapLock.RUnlock() - if len(cases) == 0 { - time.Sleep(500 * time.Millisecond) - continue - } +func (s *EventSender) printStats(cs stats, interval time.Duration) { + // initialize tabwriter - _, value, ok := reflect.Select(cases) - if !ok { - continue - } + fmt.Fprint(s.writer, "\n--------------------------------------------") + s.writer.Flush() - e := value.Interface().(*GenericEvent.Event) - <-s.process - go s.sender.SendEvent(e, s.acks, s.nacks, s.undelivereds) - } -} + fmt.Fprintf(s.writer, "\n%s\t%s\t%s\t%s", "TYPE", "ACK", "NACK", "UNDELIVERED") + fmt.Fprintf(s.writer, "\n%s\t%s\t%s\t%s\t", "----", "----", "----", "----") -func (s *EventSender) ComputeTotalEventsPerSecond() int { - eps := 0 - s.mapLock.RLock() - for _, subs := range s.events { - for _, e := range subs { - eps += e.Eps() - } + var ak []string + for k := range cs { + ak = append(ak, k) } - s.mapLock.RUnlock() - return eps -} - -func (s *EventSender) refillMaxEps(d time.Duration) { - defer func() { - if r := recover(); r != nil { - fmt.Println("Recovered in refillMaxEps", r) - } - }() - - t := time.NewTicker(d) - for { - select { - case <-t.C: - for i := 0; i < s.config.EpsLimit; i++ { - s.process <- true - } - case <-s.ctx.Done(): - return - } + sort.Strings(ak) + for _, k := range ak { + fmt.Fprintf(s.writer, "\n%v\t%v\t%v\t%v", k, float64(cs[k].acks)/interval.Seconds(), float64(cs[k].nacks)/interval.Seconds(), float64(cs[k].undelivered)/interval.Seconds()) } + s.writer.Flush() + fmt.Fprintln(s.writer, "\n--------------------------------------------") + s.writer.Flush() + fmt.Fprintf(s.writer, "%s\t%s\t%s\t", "ACK", "NACK", "UNDELIVERED") + fmt.Fprintf(s.writer, "\n%s\t%s\t%s\t", "----", "----", "----") } diff --git a/resources/loadtest/base/configmap.yaml b/resources/loadtest/base/configmap.yaml index 382e9f3..d1415df 100644 --- a/resources/loadtest/base/configmap.yaml +++ b/resources/loadtest/base/configmap.yaml @@ -4,7 +4,7 @@ metadata: name: loadtest-publisher namespace: eventing-test data: - publish_endpoint: "http://eventing-publisher-proxy.kyma-system" + publish_host: "http://eventing-publisher-proxy.kyma-system" use_legacy_events: "false" event_source: "noapp" version_format: "v%04d" diff --git a/resources/loadtest/base/kustomization.yaml b/resources/loadtest/base/kustomization.yaml index 1573173..f4216b7 100644 --- a/resources/loadtest/base/kustomization.yaml +++ b/resources/loadtest/base/kustomization.yaml @@ -17,3 +17,4 @@ resources: - rbac.yaml - service.yaml - subscriber.yaml +- subscriptions.yaml diff --git a/resources/loadtest/base/subscriptions.yaml b/resources/loadtest/base/subscriptions.yaml new file mode 100644 index 0000000..fd2f52c --- /dev/null +++ b/resources/loadtest/base/subscriptions.yaml @@ -0,0 +1,47 @@ +apiVersion: v1 +kind: List +items: + - apiVersion: eventing.kyma-project.io/v1alpha2 + kind: Subscription + metadata: + labels: + app: partner-handler-queue + eventing-loadtest: cloudevent + name: cloudevents + namespace: eventing-test + spec: + sink: http://loadtest-subscriber-0.eventing-test.svc.cluster.local + config: + maxInFlightMessages: "200" + source: "cloudevent" + typeMatching: standard + types: + - CE.v8 + - CE.v16 + - CE.v32 + - CE.v64 + - CE.v128 + - CE.v256 + - CE.v512 + - apiVersion: eventing.kyma-project.io/v1alpha2 + kind: Subscription + metadata: + labels: + app: partner-handler-queue + eventing-loadtest: legacy + name: legacy + namespace: eventing-test + spec: + sink: http://loadtest-subscriber-1.eventing-test.svc.cluster.local + config: + maxInFlightMessages: "200" + source: "legacy" + typeMatching: standard + types: + - LEGACY.v8 + - LEGACY.v16 + - LEGACY.v32 + - LEGACY.v64 + - LEGACY.v128 + - LEGACY.v256 + - LEGACY.v512