Skip to content
This repository was archived by the owner on Dec 11, 2023. It is now read-only.

Commit 2037b21

Browse files
tzununbekovtzununbekov
committed
Cleanup triggers, check target availability, combined filter fix
1 parent 2b1c910 commit 2037b21

9 files changed

Lines changed: 86 additions & 42 deletions

File tree

‎cmd/create/broker.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@ func (o *CreateOptions) broker(name string) error {
5454
}
5555

5656
log.Println("Updating manifest")
57-
restart, err := triggermesh.WriteObject(ctx, broker, path.Join(configDir, manifestFile))
57+
restart, err := triggermesh.WriteObject(broker, path.Join(configDir, manifestFile))
5858
if err != nil {
5959
return err
6060
}

‎cmd/create/source.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,7 @@ func (o *CreateOptions) source(name, kind string, args []string) error {
8383
}
8484

8585
log.Println("Updating manifest")
86-
restart, err := triggermesh.WriteObject(ctx, s, manifest)
86+
restart, err := triggermesh.WriteObject(s, manifest)
8787
if err != nil {
8888
return err
8989
}

‎cmd/create/target.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ func (o *CreateOptions) target(name, kind string, args []string, eventSourcesFil
8989
}
9090

9191
log.Println("Updating manifest")
92-
restart, err := triggermesh.WriteObject(ctx, t, manifest)
92+
restart, err := triggermesh.WriteObject(t, manifest)
9393
if err != nil {
9494
return err
9595
}

‎cmd/create/transformation.go‎

Lines changed: 46 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,16 @@ func (o *CreateOptions) NewTransformationCmd() *cobra.Command {
8383
func (o *CreateOptions) transformation(name, target, file string, eventSourcesFilter, eventTypesFilter []string) error {
8484
ctx := context.Background()
8585
configDir := path.Join(o.ConfigBase, o.Context)
86-
manifest := path.Join(configDir, manifestFile)
86+
manifestFile := path.Join(configDir, manifestFile)
87+
88+
var targetPort string
89+
if target != "" {
90+
port, err := o.lookupTarget(ctx, target)
91+
if err != nil {
92+
return err
93+
}
94+
targetPort = port
95+
}
8796

8897
for _, source := range eventSourcesFilter {
8998
et, err := o.producersEventTypes(source)
@@ -117,7 +126,7 @@ func (o *CreateOptions) transformation(name, target, file string, eventSourcesFi
117126
}
118127
}
119128
log.Println("Updating manifest")
120-
restart, err := triggermesh.WriteObject(ctx, t, manifest)
129+
restart, err := triggermesh.WriteObject(t, manifestFile)
121130
if err != nil {
122131
return err
123132
}
@@ -127,29 +136,38 @@ func (o *CreateOptions) transformation(name, target, file string, eventSourcesFi
127136
return err
128137
}
129138

130-
var targetFilters []tmbroker.Filter
139+
var targetTriggers []tmbroker.Trigger
131140
if target != "" {
132141
transformationEventTypes, err := t.GetEventTypes()
133142
if err != nil {
134143
return fmt.Errorf("transformation event type: %w", err)
135144
}
136-
if targetFilters, err = o.updateTarget(ctx, target, transformationEventTypes); err != nil {
145+
if targetTriggers, err = o.updateTarget(target, targetPort, transformationEventTypes); err != nil {
137146
return fmt.Errorf("updating target: %w", err)
138147
}
139148
}
140149

141-
var filters []tmbroker.Filter
142150
if len(eventTypesFilter) != 0 {
143-
filters = []tmbroker.Filter{tmbroker.FilterType(eventTypesFilter)}
144-
} else if len(targetFilters) != 0 {
145-
filters = targetFilters
146-
}
147-
if len(filters) != 0 {
148151
log.Println("Creating trigger")
149-
if err := o.createTrigger("", container.Name, container.HostPort(), filters...); err != nil {
152+
if err := o.createTrigger("", container.Name, container.HostPort(), []tmbroker.Filter{tmbroker.FilterType(eventTypesFilter)}...); err != nil {
150153
return err
151154
}
155+
for _, trigger := range targetTriggers {
156+
if err := trigger.RemoveTriggerFromConfig(); err != nil {
157+
return err
158+
}
159+
if err := triggermesh.RemoveObject(trigger.Name, manifestFile); err != nil {
160+
return err
161+
}
162+
}
163+
} else {
164+
for _, trigger := range targetTriggers {
165+
if err := o.createTrigger(trigger.Name, container.Name, container.HostPort(), trigger.Filters...); err != nil {
166+
return err
167+
}
168+
}
152169
}
170+
153171
output.PrintStatus("consumer", t, eventSourcesFilter, eventTypesFilter)
154172
return nil
155173
}
@@ -179,43 +197,38 @@ func readInput() (string, error) {
179197
return lines, scn.Err()
180198
}
181199

182-
func (o *CreateOptions) updateTarget(ctx context.Context, target string, transformationEventTypes []string) ([]tmbroker.Filter, error) {
200+
func (o *CreateOptions) lookupTarget(ctx context.Context, target string) (string, error) {
183201
manifest := path.Join(o.ConfigBase, o.Context, manifestFile)
184202
targetObject, err := o.getObject(target, manifest)
185203
if err != nil {
186-
return []tmbroker.Filter{}, fmt.Errorf("transformation target: %w", err)
204+
return "", fmt.Errorf("transformation target: %w", err)
187205
}
188206
consumer, ok := targetObject.(triggermesh.Consumer)
189207
if !ok {
190-
return []tmbroker.Filter{}, fmt.Errorf("%q is not an event consumer", target)
191-
}
192-
targetPort, err := consumer.GetPort(ctx)
193-
if err != nil {
194-
return []tmbroker.Filter{}, fmt.Errorf("target port: %w", err)
208+
return "", fmt.Errorf("%q is not an event consumer", target)
195209
}
210+
return consumer.GetPort(ctx)
211+
}
212+
213+
func (o *CreateOptions) updateTarget(target, targetPort string, transformationEventTypes []string) ([]tmbroker.Trigger, error) {
196214
broker, err := tmbroker.New(o.Context, path.Join(o.ConfigBase, o.Context))
197215
if err != nil {
198-
return []tmbroker.Filter{}, fmt.Errorf("broker: %w", err)
216+
return []tmbroker.Trigger{}, fmt.Errorf("broker: %w", err)
199217
}
200218
triggers, err := broker.GetTriggers()
201219
if err != nil {
202-
return []tmbroker.Filter{}, fmt.Errorf("triggers: %w", err)
220+
return []tmbroker.Trigger{}, fmt.Errorf("triggers: %w", err)
203221
}
204-
var triggerExisted bool
205-
var targetFilters []tmbroker.Filter
206-
for triggerName, triggerToTarget := range triggers {
222+
var targetTriggers []tmbroker.Trigger
223+
for name, triggerToTarget := range triggers {
207224
if target == triggerToTarget.GetTarget().Component {
208-
triggerExisted = true
209-
targetFilters = append(targetFilters, triggerToTarget.GetFilters()...)
210-
if err := o.createTrigger(triggerName, target, targetPort, tmbroker.FilterType(transformationEventTypes)); err != nil {
211-
return []tmbroker.Filter{}, fmt.Errorf("update trigger: %w", err)
212-
}
225+
triggerToTarget.Name = name
226+
triggerToTarget.BrokerConfigDir = path.Join(o.ConfigBase, o.Context)
227+
targetTriggers = append(targetTriggers, triggerToTarget)
213228
}
214229
}
215-
if !triggerExisted {
216-
if err := o.createTrigger("", target, targetPort, tmbroker.FilterType(transformationEventTypes)); err != nil {
217-
return []tmbroker.Filter{}, fmt.Errorf("create trigger: %w", err)
218-
}
230+
if err := o.createTrigger("", target, targetPort, tmbroker.FilterType(transformationEventTypes)); err != nil {
231+
return []tmbroker.Trigger{}, fmt.Errorf("create trigger: %w", err)
219232
}
220-
return targetFilters, nil
233+
return targetTriggers, nil
221234
}

‎cmd/delete/delete.go‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,7 @@ func (o *DeleteOptions) deleteComponents(components []string) error {
105105
}
106106
o.stopContainer(ctx, object.Metadata.Name, client)
107107
o.removeObject(object.Metadata.Name, currentManifest)
108+
o.cleanupTriggers(object.Metadata.Name, currentManifest)
108109
}
109110
return currentManifest.Write()
110111
}
@@ -128,3 +129,24 @@ func (o *DeleteOptions) removeObject(component string, manifest *manifest.Manife
128129
func (o *DeleteOptions) stopContainer(ctx context.Context, name string, client *client.Client) error {
129130
return docker.ForceStop(ctx, name, client)
130131
}
132+
133+
func (o *DeleteOptions) cleanupTriggers(component string, manifest *manifest.Manifest) {
134+
for _, object := range manifest.Objects {
135+
if object.Kind != "Trigger" {
136+
continue
137+
}
138+
trigger := tmbroker.NewTrigger(object.Metadata.Name, o.Context, path.Join(o.ConfigDir, o.Context))
139+
if err := trigger.LookupTrigger(); err != nil {
140+
log.Printf("Trigger config %q: %v", object.Metadata.Name, err)
141+
continue
142+
}
143+
if trigger.Target.Component != component {
144+
continue
145+
}
146+
if err := trigger.RemoveTriggerFromConfig(); err != nil {
147+
log.Printf("Deleting trigger %q: %v", object.Metadata.Name, err)
148+
continue
149+
}
150+
manifest.Remove(object.Metadata.Name)
151+
}
152+
}

‎cmd/start/start.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,7 @@ func (o *StartOptions) start(broker string) error {
105105
} else {
106106
componentTriggers[trigger.GetTarget().Component] = []*tmbroker.Trigger{trigger}
107107
}
108-
if _, err := triggermesh.WriteObject(ctx, trigger, manifestFile); err != nil {
108+
if _, err := triggermesh.WriteObject(trigger, manifestFile); err != nil {
109109
return fmt.Errorf("creating trigger: %w", err)
110110
}
111111
case "Secret":
@@ -161,7 +161,7 @@ func (o *StartOptions) start(broker string) error {
161161
if c == nil {
162162
continue
163163
}
164-
if _, err := triggermesh.WriteObject(ctx, c.(triggermesh.Component), manifestFile); err != nil {
164+
if _, err := triggermesh.WriteObject(c.(triggermesh.Component), manifestFile); err != nil {
165165
return fmt.Errorf("updating manifest: %w", err)
166166
}
167167
}

‎pkg/output/output.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -172,7 +172,7 @@ func triggerFilterToString(filter tmbroker.Filter) string {
172172
if err != nil {
173173
return ""
174174
}
175-
result := strings.ReplaceAll(string(f), "\n", " ")
175+
result := strings.ReplaceAll(string(f), "\n", ":")
176176
result = strings.ReplaceAll(result, " ", "")
177177
result = strings.ReplaceAll(result, ":", " ")
178178
return result

‎pkg/triggermesh/components/broker/trigger.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -176,7 +176,7 @@ func (t *Trigger) UpdateBrokerConfig() error {
176176
}
177177

178178
func (t *Trigger) UpdateManifest() error {
179-
m := manifest.New(path.Join(t.BrokerConfigDir, "manifest.yaml"))
179+
m := manifest.New(path.Join(t.BrokerConfigDir, manifestFile))
180180
if err := m.Read(); err != nil {
181181
return fmt.Errorf("manifest read: %w", err)
182182
}

‎pkg/triggermesh/triggermesh.go‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ import (
2626
"github.com/triggermesh/tmcli/pkg/manifest"
2727
)
2828

29-
func WriteObject(ctx context.Context, object Component, manifestFile string) (bool, error) {
29+
func WriteObject(object Component, manifestFile string) (bool, error) {
3030
manifest := manifest.New(manifestFile)
3131
if err := manifest.Read(); err != nil {
3232
return false, fmt.Errorf("reading manifest: %w", err)
@@ -41,6 +41,15 @@ func WriteObject(ctx context.Context, object Component, manifestFile string) (bo
4141
return true, manifest.Write()
4242
}
4343

44+
func RemoveObject(name string, manifestFile string) error {
45+
manifest := manifest.New(manifestFile)
46+
if err := manifest.Read(); err != nil {
47+
return fmt.Errorf("reading manifest: %w", err)
48+
}
49+
manifest.Remove(name)
50+
return manifest.Write()
51+
}
52+
4453
func Start(ctx context.Context, object Runnable, restart bool, additionalEnv map[string]string) (*docker.Container, error) {
4554
client, err := docker.NewClient()
4655
if err != nil {
@@ -101,7 +110,7 @@ func ProcessSecrets(ctx context.Context, p Parent, manifestFile string) (map[str
101110
secretsChanged := false
102111
secretEnv := make(map[string]string)
103112
for _, s := range secrets {
104-
dirty, err := WriteObject(ctx, s, manifestFile)
113+
dirty, err := WriteObject(s, manifestFile)
105114
if err != nil {
106115
return nil, false, fmt.Errorf("write nested object: %w", err)
107116
}

0 commit comments

Comments
 (0)