983758645

Committed 28 Aug 2023 09:13AM UTC coverage: 83.45%. First build

Build Type

gitlab-ci

Committed by Mender

Commit Message

fix: `migrate` command tries to create Jetstream consumer before stream

Changelog: Title
Ticket: MEN-6698
Signed-off-by: Alf-Rune Siqveland <alf.rune@northern.tech>
(cherry picked from commit 27e9e8416)

Pull Request Pull Request #282: [Cherry 2.5.x]: fix: `migrate` command tries to create Jetstream consumer before stream

Run Details

0 of 2 new or added lines in 1 file covered. (0.0%)

1553 of 1861 relevant lines covered (83.45%)

14.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.71

/main.go

// Copyright 2023 Northern.tech AS
//
//    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 (
        "context"
        "fmt"
        "log"
        "os"
        "strings"

        "github.com/mendersoftware/go-lib-micro/config"
        "github.com/pkg/errors"
        "github.com/urfave/cli"

        "github.com/mendersoftware/workflows/app/server"
        "github.com/mendersoftware/workflows/app/worker"
        "github.com/mendersoftware/workflows/client/nats"
        dconfig "github.com/mendersoftware/workflows/config"
        "github.com/mendersoftware/workflows/model"
        store "github.com/mendersoftware/workflows/store/mongo"
)

func main() {
        doMain(os.Args)
}

func doMain(args []string) {
        var configPath string

        app := &cli.App{
                Flags: []cli.Flag{
                        &cli.StringFlag{
                                Name: "config",
                                Usage: "Configuration `FILE`." +
                                        " Supports JSON, TOML, YAML and HCL formatted configs.",
                                Destination: &configPath,
                        },
                },
                Commands: []cli.Command{
                        {
                                Name:   "server",
                                Usage:  "Run the HTTP API server",
                                Action: cmdServer,
                                Flags: []cli.Flag{
                                        &cli.BoolFlag{
                                                Name:  "automigrate",
                                                Usage: "Run database migrations before starting.",
                                        },
                                },
                        },
                        {
                                Name:   "worker",
                                Usage:  "Run the worker process",
                                Action: cmdWorker,
                                Flags: []cli.Flag{
                                        &cli.BoolFlag{
                                                Name:  "automigrate",
                                                Usage: "Run database migrations before starting.",
                                        },
                                        &cli.StringFlag{
                                                Name:  "workflows",
                                                Usage: "Comma-separated list of workflows executed by this worker",
                                        },
                                        &cli.StringFlag{
                                                Name:  "excluded-workflows",
                                                Usage: "Comma-separated list of workflows NOT executed by this worker",
                                        },
                                },
                        },
                        {
                                Name:   "migrate",
                                Usage:  "Run the migrations",
                                Action: cmdMigrate,
                                Flags: []cli.Flag{
                                        cli.BoolFlag{
                                                Name:   "skip-nats",
                                                Usage:  "Skip migrating the NATS Jetstream configuration",
                                                EnvVar: "WORKFLOWS_MIGRATION_SKIP_NATS",
                                        },
                                        cli.BoolFlag{
                                                Name:   "skip-database",
                                                Usage:  "Skip migrating the database",
                                                EnvVar: "WORKFLOWS_MIGRATION_SKIP_DATABASE",
                                        },
                                },
                        },
                        {
                                Name:   "list-jobs",
                                Usage:  "List jobs",
                                Action: cmdListJobs,
                                Flags: []cli.Flag{
                                        cli.Int64Flag{
                                                Name:  "page",
                                                Usage: "page number to show",
                                        },
                                        cli.Int64Flag{
                                                Name:  "perPage",
                                                Usage: "number of results per page",
                                        },
                                },
                        },
                },
        }
        app.Usage = "Workflows"
        app.Action = cmdServer

        app.Before = func(args *cli.Context) error {
                err := config.FromConfigFile(configPath, dconfig.Defaults)
                if err != nil {
                        return cli.NewExitError(
                                fmt.Sprintf("error loading configuration: %s", err),
                                1)
                }

                // Enable setting config values by environment variables
                config.Config.SetEnvPrefix("WORKFLOWS")
                config.Config.AutomaticEnv()
                config.Config.SetEnvKeyReplacer(strings.NewReplacer(".", "_", "-", "_"))

                return nil
        }

        err := app.Run(args)
        if err != nil {
                log.Fatal(err)
        }
}

func getNatsClient() (nats.Client, error) {
        natsURI := config.Config.GetString(dconfig.SettingNatsURI)
        streamName := config.Config.GetString(dconfig.SettingNatsStreamName)
        nats, err := nats.NewClientWithDefaults(natsURI, streamName)
        if err != nil {
                return nil, errors.Wrap(err, "failed to connect to nats")
        }
        return nats, err
}

func initJetstream(nc nats.Client, producer, upsert bool) (err error) {
        durableName := config.Config.GetString(dconfig.SettingNatsSubscriberDurable)
        if producer {
                err = nc.JetStreamCreateStream(nc.StreamName())
                if err != nil {
                        return err
                }
        } else {
                var cfg nats.ConsumerConfig
                cfg, err = dconfig.GetNatsConsumerConfig(config.Config)
                if err != nil {
                        return err
                }
                err = nc.CreateConsumer(durableName, upsert, cfg)
        }
        return err
}

func cmdServer(args *cli.Context) error {
        dataStore, err := store.SetupDataStore(args.Bool("automigrate"))
        if err != nil {
                return err
        }
        defer dataStore.Close()

        nats, err := getNatsClient()
        if err != nil {
                return err
        }
        defer nats.Close()

        if err = initJetstream(nats, true, args.Bool("automigrate")); err != nil {
                return errors.WithMessage(err, "failed to apply Jetstream migrations")
        }

        return server.InitAndRun(config.Config, dataStore, nats)
}

func cmdWorker(args *cli.Context) error {
        dataStore, err := store.SetupDataStore(args.Bool("automigrate"))
        if err != nil {
                return err
        }
        defer dataStore.Close()

        nats, err := getNatsClient()
        if err != nil {
                return err
        }
        defer nats.Close()

        if err = initJetstream(nats, false, args.Bool("automigrate")); err != nil {
                return errors.WithMessage(err, "failed to apply Jetstream consumer migrations")
        }

        var included, excluded []string

        includedWorkflows := args.String("workflows")
        if includedWorkflows != "" {
                included = strings.Split(includedWorkflows, ",")
        }

        excludedWorkflows := args.String("excluded-workflows")
        if excludedWorkflows != "" {
                excluded = strings.Split(excludedWorkflows, ",")
        }

        workflows := worker.Workflows{
                Included: included,
                Excluded: excluded,
        }
        return worker.InitAndRun(config.Config, workflows, dataStore, nats)
}

func cmdMigrate(args *cli.Context) error {
        var err error
        if !args.Bool("skip-database") {
                _, err = store.SetupDataStore(true)
                if err != nil {
                        return err
                }
        }
        if !args.Bool("skip-nats") {
                var nc nats.Client
                nc, err = getNatsClient()
                if err == nil {
                        if err = initJetstream(nc, true, true); err == nil {
                                err = initJetstream(nc, false, true)
                        }
                }

        }
        return err
}

func cmdListJobs(args *cli.Context) error {
        dataStore, err := store.SetupDataStore(false)
        if err != nil {
                return err
        }
        defer dataStore.Close()

        ctx, cancel := context.WithCancel(context.Background())
        defer cancel()

        var page int64
        var perPage int64
        page = args.Int64("page")
        perPage = args.Int64("perPage")

        if page < 1 {
                page = 1
        }
        if perPage < 1 {
                perPage = 75
        }
        jobs, count, _ := dataStore.GetAllJobs(ctx, page, perPage)
        fmt.Printf("all jobs: %d; page: %d/%d perPage:%d\n%29s %24s %10s %s\n",
                count, page, count/perPage, perPage, "insert time", "id", "status", "workflow")
        for _, j := range jobs {
                format := "Mon, 2 Jan 2006 15:04:05 MST"
                fmt.Printf(
                        "%29s %24s %10s %s\n",
                        j.InsertTime.Format(format),
                        j.ID, model.StatusToString(j.Status),
                        j.WorkflowName,
                )
        }
        fmt.Printf("all jobs: %d; page: %d/%d\n", count, page, count/perPage)

        return nil
}

1	// Copyright 2023 Northern.tech AS
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	package main
16
17	import (
18	"context"
19	"fmt"
20	"log"
21	"os"
22	"strings"
23
24	"github.com/mendersoftware/go-lib-micro/config"
25	"github.com/pkg/errors"
26	"github.com/urfave/cli"
27
28	"github.com/mendersoftware/workflows/app/server"
29	"github.com/mendersoftware/workflows/app/worker"
30	"github.com/mendersoftware/workflows/client/nats"
31	dconfig "github.com/mendersoftware/workflows/config"
32	"github.com/mendersoftware/workflows/model"
33	store "github.com/mendersoftware/workflows/store/mongo"
34	)
35
36	func main() {	×
37	doMain(os.Args)	×
38	}	×
39
40	func doMain(args []string) {	2✔
41	var configPath string	2✔
42		2✔
43	app := &cli.App{	2✔
44	Flags: []cli.Flag{	2✔
45	&cli.StringFlag{	2✔
46	Name: "config",	2✔
47	Usage: "Configuration `FILE`." +	2✔
48	" Supports JSON, TOML, YAML and HCL formatted configs.",	2✔
49	Destination: &configPath,	2✔
50	},	2✔
51	},	2✔
52	Commands: []cli.Command{	2✔
53	{	2✔
54	Name: "server",	2✔
55	Usage: "Run the HTTP API server",	2✔
56	Action: cmdServer,	2✔
57	Flags: []cli.Flag{	2✔
58	&cli.BoolFlag{	2✔
59	Name: "automigrate",	2✔
60	Usage: "Run database migrations before starting.",	2✔
61	},	2✔
62	},	2✔
63	},	2✔
64	{	2✔
65	Name: "worker",	2✔
66	Usage: "Run the worker process",	2✔
67	Action: cmdWorker,	2✔
68	Flags: []cli.Flag{	2✔
69	&cli.BoolFlag{	2✔
70	Name: "automigrate",	2✔
71	Usage: "Run database migrations before starting.",	2✔
72	},	2✔
73	&cli.StringFlag{	2✔
74	Name: "workflows",	2✔
75	Usage: "Comma-separated list of workflows executed by this worker",	2✔
76	},	2✔
77	&cli.StringFlag{	2✔
78	Name: "excluded-workflows",	2✔
79	Usage: "Comma-separated list of workflows NOT executed by this worker",	2✔
80	},	2✔
81	},	2✔
82	},	2✔
83	{	2✔
84	Name: "migrate",	2✔
85	Usage: "Run the migrations",	2✔
86	Action: cmdMigrate,	2✔
87	Flags: []cli.Flag{	2✔
88	cli.BoolFlag{	2✔
89	Name: "skip-nats",	2✔
90	Usage: "Skip migrating the NATS Jetstream configuration",	2✔
91	EnvVar: "WORKFLOWS_MIGRATION_SKIP_NATS",	2✔
92	},	2✔
93	cli.BoolFlag{	2✔
94	Name: "skip-database",	2✔
95	Usage: "Skip migrating the database",	2✔
96	EnvVar: "WORKFLOWS_MIGRATION_SKIP_DATABASE",	2✔
97	},	2✔
98	},	2✔
99	},	2✔
100	{	2✔
101	Name: "list-jobs",	2✔
102	Usage: "List jobs",	2✔
103	Action: cmdListJobs,	2✔
104	Flags: []cli.Flag{	2✔
105	cli.Int64Flag{	2✔
106	Name: "page",	2✔
107	Usage: "page number to show",	2✔
108	},	2✔
109	cli.Int64Flag{	2✔
110	Name: "perPage",	2✔
111	Usage: "number of results per page",	2✔
112	},	2✔
113	},	2✔
114	},	2✔
115	},	2✔
116	}	2✔
117	app.Usage = "Workflows"	2✔
118	app.Action = cmdServer	2✔
119		2✔
120	app.Before = func(args *cli.Context) error {	4✔
121	err := config.FromConfigFile(configPath, dconfig.Defaults)	2✔
122	if err != nil {	2✔
123	return cli.NewExitError(	×
124	fmt.Sprintf("error loading configuration: %s", err),	×
125	1)	×
126	}	×
127
128	// Enable setting config values by environment variables
129	config.Config.SetEnvPrefix("WORKFLOWS")	2✔
130	config.Config.AutomaticEnv()	2✔
131	config.Config.SetEnvKeyReplacer(strings.NewReplacer(".", "_", "-", "_"))	2✔
132		2✔
133	return nil	2✔
134	}
135
136	err := app.Run(args)	2✔
137	if err != nil {	2✔
138	log.Fatal(err)	×
139	}	×
140	}
141
142	func getNatsClient() (nats.Client, error) {	2✔
143	natsURI := config.Config.GetString(dconfig.SettingNatsURI)	2✔
144	streamName := config.Config.GetString(dconfig.SettingNatsStreamName)	2✔
145	nats, err := nats.NewClientWithDefaults(natsURI, streamName)	2✔
146	if err != nil {	2✔
147	return nil, errors.Wrap(err, "failed to connect to nats")	×
148	}	×
149	return nats, err	2✔
150	}
151
152	func initJetstream(nc nats.Client, producer, upsert bool) (err error) {	2✔
153	durableName := config.Config.GetString(dconfig.SettingNatsSubscriberDurable)	2✔
154	if producer {	3✔
155	err = nc.JetStreamCreateStream(nc.StreamName())	1✔
156	if err != nil {	1✔
157	return err	×
158	}	×
159	} else {	1✔
160	var cfg nats.ConsumerConfig	1✔
161	cfg, err = dconfig.GetNatsConsumerConfig(config.Config)	1✔
162	if err != nil {	1✔
163	return err	×
164	}	×
165	err = nc.CreateConsumer(durableName, upsert, cfg)	1✔
166	}
167	return err	2✔
168	}
169
170	func cmdServer(args *cli.Context) error {	1✔
171	dataStore, err := store.SetupDataStore(args.Bool("automigrate"))	1✔
172	if err != nil {	1✔
173	return err	×
174	}	×
175	defer dataStore.Close()	1✔
176		1✔
177	nats, err := getNatsClient()	1✔
178	if err != nil {	1✔
179	return err	×
180	}	×
181	defer nats.Close()	1✔
182		1✔
183	if err = initJetstream(nats, true, args.Bool("automigrate")); err != nil {	1✔
184	return errors.WithMessage(err, "failed to apply Jetstream migrations")	×
185	}	×
186
187	return server.InitAndRun(config.Config, dataStore, nats)	1✔
188	}
189
190	func cmdWorker(args *cli.Context) error {	1✔
191	dataStore, err := store.SetupDataStore(args.Bool("automigrate"))	1✔
192	if err != nil {	1✔
193	return err	×
194	}	×
195	defer dataStore.Close()	1✔
196		1✔
197	nats, err := getNatsClient()	1✔
198	if err != nil {	1✔
199	return err	×
200	}	×
201	defer nats.Close()	1✔
202		1✔
203	if err = initJetstream(nats, false, args.Bool("automigrate")); err != nil {	1✔
204	return errors.WithMessage(err, "failed to apply Jetstream consumer migrations")	×
205	}	×
206
207	var included, excluded []string	1✔
208		1✔
209	includedWorkflows := args.String("workflows")	1✔
210	if includedWorkflows != "" {	2✔
211	included = strings.Split(includedWorkflows, ",")	1✔
212	}	1✔
213
214	excludedWorkflows := args.String("excluded-workflows")	1✔
215	if excludedWorkflows != "" {	2✔
216	excluded = strings.Split(excludedWorkflows, ",")	1✔
217	}	1✔
218
219	workflows := worker.Workflows{	1✔
220	Included: included,	1✔
221	Excluded: excluded,	1✔
222	}	1✔
223	return worker.InitAndRun(config.Config, workflows, dataStore, nats)	1✔
224	}
225
226	func cmdMigrate(args *cli.Context) error {	×
227	var err error	×
228	if !args.Bool("skip-database") {	×
229	_, err = store.SetupDataStore(true)	×
230	if err != nil {	×
231	return err	×
232	}	×
233	}
234	if !args.Bool("skip-nats") {	×
235	var nc nats.Client	×
236	nc, err = getNatsClient()	×
237	if err == nil {	×
NEW 238	if err = initJetstream(nc, true, true); err == nil {	×
NEW 239	err = initJetstream(nc, false, true)	×
240	}	×
241	}
242
243	}
244	return err	×
245	}
246
247	func cmdListJobs(args *cli.Context) error {	×
248	dataStore, err := store.SetupDataStore(false)	×
249	if err != nil {	×
250	return err	×
251	}	×
252	defer dataStore.Close()	×
253		×
254	ctx, cancel := context.WithCancel(context.Background())	×
255	defer cancel()	×
256		×
257	var page int64	×
258	var perPage int64	×
259	page = args.Int64("page")	×
260	perPage = args.Int64("perPage")	×
261		×
262	if page < 1 {	×
263	page = 1	×
264	}	×
265	if perPage < 1 {	×
266	perPage = 75	×
267	}	×
268	jobs, count, _ := dataStore.GetAllJobs(ctx, page, perPage)	×
269	fmt.Printf("all jobs: %d; page: %d/%d perPage:%d\n%29s %24s %10s %s\n",	×
270	count, page, count/perPage, perPage, "insert time", "id", "status", "workflow")	×
271	for _, j := range jobs {	×
272	format := "Mon, 2 Jan 2006 15:04:05 MST"	×
273	fmt.Printf(	×
274	"%29s %24s %10s %s\n",	×
275	j.InsertTime.Format(format),	×
276	j.ID, model.StatusToString(j.Status),	×
277	j.WorkflowName,	×
278	)	×
279	}	×
280	fmt.Printf("all jobs: %d; page: %d/%d\n", count, page, count/perPage)	×
281		×
282	return nil	×
283	}

mendersoftware / workflows / 983758645

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous