1592525883

Committed 17 Dec 2024 01:50PM UTC coverage: 73.526% (+0.7%) from 72.839%

Build # 1592525883

Build Type

Pull #270

gitlab-ci

Committed by

bahaa-ghazal

Commit Message

test: testing 'deploy to all devices' feature

Changelog = Title
Ticket = MEN-4272
Signed-off-by: Bahaa Aldeen Ghazal <bahaa.ghazal@northern.tech>

Pull Request Pull Request #270: test: testing 'deploy to all devices' feature

Run Details

4244 of 6144 branches covered (69.08%)

Branch coverage included in aggregate %.

40043 of 54089 relevant lines covered (74.03%)

23.11 hits per line

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

63.45

/backend/services/workflows/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"
        "encoding/json"
        "fmt"
        "log"
        "os"
        "strings"

        "github.com/pkg/errors"
        "github.com/urfave/cli"

        "github.com/mendersoftware/mender-server/pkg/config"
        "github.com/mendersoftware/mender-server/pkg/version"

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

var appVersion = version.Get()

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",
                                        },
                                },
                        },
                        {
                                Name:  "version",
                                Usage: "Show version information",
                                Flags: []cli.Flag{
                                        cli.StringFlag{
                                                Name:  "output",
                                                Usage: "Output format <json|text>",
                                                Value: "text",
                                        },
                                },
                                Action: func(args *cli.Context) error {
                                        switch strings.ToLower(args.String("output")) {
                                        case "text":
                                                fmt.Print(appVersion)
                                        case "json":
                                                _ = json.NewEncoder(os.Stdout).Encode(appVersion)
                                        default:
                                                return fmt.Errorf("Unknown output format %q", args.String("output"))
                                        }
                                        return nil
                                },
                        },
                },
                Version: appVersion.Version,
        }
        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	"encoding/json"
20	"fmt"
21	"log"
22	"os"
23	"strings"
24
25	"github.com/pkg/errors"
26	"github.com/urfave/cli"
27
28	"github.com/mendersoftware/mender-server/pkg/config"
29	"github.com/mendersoftware/mender-server/pkg/version"
30
31	"github.com/mendersoftware/mender-server/services/workflows/app/server"
32	"github.com/mendersoftware/mender-server/services/workflows/app/worker"
33	"github.com/mendersoftware/mender-server/services/workflows/client/nats"
34	dconfig "github.com/mendersoftware/mender-server/services/workflows/config"
35	"github.com/mendersoftware/mender-server/services/workflows/model"
36	store "github.com/mendersoftware/mender-server/services/workflows/store/mongo"
37	)
38
39	var appVersion = version.Get()
40
41	func main() {	2✔
42	doMain(os.Args)	2✔
43	}	2✔
44
45	func doMain(args []string) {	2✔
46	var configPath string	2✔
47		2✔
48	app := &cli.App{	2✔
49	Flags: []cli.Flag{	2✔
50	&cli.StringFlag{	2✔
51	Name: "config",	2✔
52	Usage: "Configuration `FILE`." +	2✔
53	" Supports JSON, TOML, YAML and HCL formatted configs.",	2✔
54	Destination: &configPath,	2✔
55	},	2✔
56	},	2✔
57	Commands: []cli.Command{	2✔
58	{	2✔
59	Name: "server",	2✔
60	Usage: "Run the HTTP API server",	2✔
61	Action: cmdServer,	2✔
62	Flags: []cli.Flag{	2✔
63	&cli.BoolFlag{	2✔
64	Name: "automigrate",	2✔
65	Usage: "Run database migrations before starting.",	2✔
66	},	2✔
67	},	2✔
68	},	2✔
69	{	2✔
70	Name: "worker",	2✔
71	Usage: "Run the worker process",	2✔
72	Action: cmdWorker,	2✔
73	Flags: []cli.Flag{	2✔
74	&cli.BoolFlag{	2✔
75	Name: "automigrate",	2✔
76	Usage: "Run database migrations before starting.",	2✔
77	},	2✔
78	&cli.StringFlag{	2✔
79	Name: "workflows",	2✔
80	Usage: "Comma-separated list of workflows executed by this worker",	2✔
81	},	2✔
82	&cli.StringFlag{	2✔
83	Name: "excluded-workflows",	2✔
84	Usage: "Comma-separated list of workflows NOT executed by this worker",	2✔
85	},	2✔
86	},	2✔
87	},	2✔
88	{	2✔
89	Name: "migrate",	2✔
90	Usage: "Run the migrations",	2✔
91	Action: cmdMigrate,	2✔
92	Flags: []cli.Flag{	2✔
93	cli.BoolFlag{	2✔
94	Name: "skip-nats",	2✔
95	Usage: "Skip migrating the NATS Jetstream configuration",	2✔
96	EnvVar: "WORKFLOWS_MIGRATION_SKIP_NATS",	2✔
97	},	2✔
98	cli.BoolFlag{	2✔
99	Name: "skip-database",	2✔
100	Usage: "Skip migrating the database",	2✔
101	EnvVar: "WORKFLOWS_MIGRATION_SKIP_DATABASE",	2✔
102	},	2✔
103	},	2✔
104	},	2✔
105	{	2✔
106	Name: "list-jobs",	2✔
107	Usage: "List jobs",	2✔
108	Action: cmdListJobs,	2✔
109	Flags: []cli.Flag{	2✔
110	cli.Int64Flag{	2✔
111	Name: "page",	2✔
112	Usage: "page number to show",	2✔
113	},	2✔
114	cli.Int64Flag{	2✔
115	Name: "perPage",	2✔
116	Usage: "number of results per page",	2✔
117	},	2✔
118	},	2✔
119	},	2✔
120	{	2✔
121	Name: "version",	2✔
122	Usage: "Show version information",	2✔
123	Flags: []cli.Flag{	2✔
124	cli.StringFlag{	2✔
125	Name: "output",	2✔
126	Usage: "Output format <json\|text>",	2✔
127	Value: "text",	2✔
128	},	2✔
129	},	2✔
130	Action: func(args *cli.Context) error {	2✔
131	switch strings.ToLower(args.String("output")) {	×
132	case "text":	×
133	fmt.Print(appVersion)	×
134	case "json":	×
135	_ = json.NewEncoder(os.Stdout).Encode(appVersion)	×
136	default:	×
137	return fmt.Errorf("Unknown output format %q", args.String("output"))	×
138	}
139	return nil	×
140	},
141	},
142	},
143	Version: appVersion.Version,
144	}
145	app.Usage = "Workflows"	2✔
146	app.Action = cmdServer	2✔
147		2✔
148	app.Before = func(args *cli.Context) error {	4✔
149	err := config.FromConfigFile(configPath, dconfig.Defaults)	2✔
150	if err != nil {	2✔
151	return cli.NewExitError(	×
152	fmt.Sprintf("error loading configuration: %s", err),	×
153	1)	×
154	}	×
155
156	// Enable setting config values by environment variables
157	config.Config.SetEnvPrefix("WORKFLOWS")	2✔
158	config.Config.AutomaticEnv()	2✔
159	config.Config.SetEnvKeyReplacer(strings.NewReplacer(".", "_", "-", "_"))	2✔
160		2✔
161	return nil	2✔
162	}
163
164	err := app.Run(args)	2✔
165	if err != nil {	2✔
166	log.Fatal(err)	×
167	}	×
168	}
169
170	func getNatsClient() (nats.Client, error) {	2✔
171	natsURI := config.Config.GetString(dconfig.SettingNatsURI)	2✔
172	streamName := config.Config.GetString(dconfig.SettingNatsStreamName)	2✔
173	nats, err := nats.NewClientWithDefaults(natsURI, streamName)	2✔
174	if err != nil {	2✔
175	return nil, errors.Wrap(err, "failed to connect to nats")	×
176	}	×
177	return nats, err	2✔
178	}
179
180	func initJetstream(nc nats.Client, producer, upsert bool) (err error) {	2✔
181	durableName := config.Config.GetString(dconfig.SettingNatsSubscriberDurable)	2✔
182	if producer {	4✔
183	err = nc.JetStreamCreateStream(nc.StreamName())	2✔
184	if err != nil {	2✔
185	return err	×
186	}	×
187	} else {	2✔
188	var cfg nats.ConsumerConfig	2✔
189	cfg, err = dconfig.GetNatsConsumerConfig(config.Config)	2✔
190	if err != nil {	2✔
191	return err	×
192	}	×
193	err = nc.CreateConsumer(durableName, upsert, cfg)	2✔
194	}
195	return err	2✔
196	}
197
198	func cmdServer(args *cli.Context) error {	2✔
199	dataStore, err := store.SetupDataStore(args.Bool("automigrate"))	2✔
200	if err != nil {	2✔
201	return err	×
202	}	×
203	defer dataStore.Close()	2✔
204		2✔
205	nats, err := getNatsClient()	2✔
206	if err != nil {	2✔
207	return err	×
208	}	×
209	defer nats.Close()	2✔
210		2✔
211	if err = initJetstream(nats, true, args.Bool("automigrate")); err != nil {	2✔
212	return errors.WithMessage(err, "failed to apply Jetstream migrations")	×
213	}	×
214
215	return server.InitAndRun(config.Config, dataStore, nats)	2✔
216	}
217
218	func cmdWorker(args *cli.Context) error {	2✔
219	dataStore, err := store.SetupDataStore(args.Bool("automigrate"))	2✔
220	if err != nil {	2✔
221	return err	×
222	}	×
223	defer dataStore.Close()	2✔
224		2✔
225	nats, err := getNatsClient()	2✔
226	if err != nil {	2✔
227	return err	×
228	}	×
229	defer nats.Close()	2✔
230		2✔
231	if err = initJetstream(nats, false, args.Bool("automigrate")); err != nil {	2✔
232	return errors.WithMessage(err, "failed to apply Jetstream consumer migrations")	×
233	}	×
234
235	var included, excluded []string	2✔
236		2✔
237	includedWorkflows := args.String("workflows")	2✔
238	if includedWorkflows != "" {	2✔
239	included = strings.Split(includedWorkflows, ",")	×
240	}	×
241
242	excludedWorkflows := args.String("excluded-workflows")	2✔
243	if excludedWorkflows != "" {	2✔
244	excluded = strings.Split(excludedWorkflows, ",")	×
245	}	×
246
247	workflows := worker.Workflows{	2✔
248	Included: included,	2✔
249	Excluded: excluded,	2✔
250	}	2✔
251	return worker.InitAndRun(config.Config, workflows, dataStore, nats)	2✔
252	}
253
254	func cmdMigrate(args *cli.Context) error {	×
255	var err error	×
256	if !args.Bool("skip-database") {	×
257	_, err = store.SetupDataStore(true)	×
258	if err != nil {	×
259	return err	×
260	}	×
261	}
262	if !args.Bool("skip-nats") {	×
263	var nc nats.Client	×
264	nc, err = getNatsClient()	×
265	if err == nil {	×
266	if err = initJetstream(nc, true, true); err == nil {	×
267	err = initJetstream(nc, false, true)	×
268	}	×
269	}
270
271	}
272	return err	×
273	}
274
275	func cmdListJobs(args *cli.Context) error {	×
276	dataStore, err := store.SetupDataStore(false)	×
277	if err != nil {	×
278	return err	×
279	}	×
280	defer dataStore.Close()	×
281		×
282	ctx, cancel := context.WithCancel(context.Background())	×
283	defer cancel()	×
284		×
285	var page int64	×
286	var perPage int64	×
287	page = args.Int64("page")	×
288	perPage = args.Int64("perPage")	×
289		×
290	if page < 1 {	×
291	page = 1	×
292	}	×
293	if perPage < 1 {	×
294	perPage = 75	×
295	}	×
296	jobs, count, _ := dataStore.GetAllJobs(ctx, page, perPage)	×
297	fmt.Printf("all jobs: %d; page: %d/%d perPage:%d\n%29s %24s %10s %s\n",	×
298	count, page, count/perPage, perPage, "insert time", "id", "status", "workflow")	×
299	for _, j := range jobs {	×
300	format := "Mon, 2 Jan 2006 15:04:05 MST"	×
301	fmt.Printf(	×
302	"%29s %24s %10s %s\n",	×
303	j.InsertTime.Format(format),	×
304	j.ID, model.StatusToString(j.Status),	×
305	j.WorkflowName,	×
306	)	×
307	}	×
308	fmt.Printf("all jobs: %d; page: %d/%d\n", count, page, count/perPage)	×
309		×
310	return nil	×
311	}

mendersoftware / mender-server / 1592525883

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