792065958

Build # 792065958

Build Type

gitlab-ci

Committed by GitHub

Commit Message

chore: bump golang.org/x/sys

Pull Request Pull Request #263: chore: bump golang.org/x/sys from 0.0.0-20220715151400-c0bba94af5f8 to 0.5.0

Run Details

1474 of 1729 relevant lines covered (85.25%)

20.29 hits per line

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

82.75

/store/mongo/datastore_mongo.go

// Copyright 2022 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 mongo

import (
        "context"
        "crypto/tls"
        "fmt"
        "strconv"
        "strings"
        "time"

        "github.com/pkg/errors"
        "go.mongodb.org/mongo-driver/bson"
        "go.mongodb.org/mongo-driver/bson/primitive"
        "go.mongodb.org/mongo-driver/mongo"
        "go.mongodb.org/mongo-driver/mongo/options"
        mopts "go.mongodb.org/mongo-driver/mongo/options"

        "github.com/mendersoftware/go-lib-micro/config"
        "github.com/mendersoftware/go-lib-micro/log"

        dconfig "github.com/mendersoftware/workflows/config"
        "github.com/mendersoftware/workflows/model"
        "github.com/mendersoftware/workflows/store"
)

const (
        // JobsCollectionName refers to the collection of finished or
        // jobs in progress.
        JobsCollectionName = "jobs"

        // WorkflowCollectionName refers to the collection of stored workflows
        WorkflowCollectionName = "workflows"
)

var (
        ErrNoSuchWorkflowByVersion = errors.New("Workflow of the given version not found")
)

// SetupDataStore returns the mongo data store and optionally runs migrations
func SetupDataStore(automigrate bool) (*DataStoreMongo, error) {
        ctx := context.Background()
        dbClient, err := NewClient(ctx, config.Config)
        if err != nil {
                return nil, errors.New(fmt.Sprintf("failed to connect to db: %v", err))
        }
        err = doMigrations(ctx, dbClient, automigrate)
        if err != nil {
                return nil, err
        }
        dataStore := NewDataStoreWithClient(dbClient, config.Config)
        return dataStore, nil
}

func doMigrations(ctx context.Context, client *Client,
        automigrate bool) error {
        db := config.Config.GetString(dconfig.SettingDbName)
        err := Migrate(ctx, db, DbVersion, client, automigrate)
        if err != nil {
                return errors.New(fmt.Sprintf("failed to run migrations: %v", err))
        }

        return nil
}

func disconnectClient(parentCtx context.Context, client *Client) {
        ctx, cancel := context.WithTimeout(parentCtx, 10*time.Second)
        defer cancel()
        _ = client.Disconnect(ctx)
}

// Client is a package specific mongo client
type Client struct {
        mongo.Client
}

// NewClient returns a mongo client
func NewClient(_ context.Context, c config.Reader) (*Client, error) {

        clientOptions := mopts.Client()
        mongoURL := c.GetString(dconfig.SettingMongo)
        if !strings.Contains(mongoURL, "://") {
                return nil, errors.Errorf("Invalid mongoURL %q: missing schema.",
                        mongoURL)
        }
        clientOptions.ApplyURI(mongoURL)

        username := c.GetString(dconfig.SettingDbUsername)
        if username != "" {
                credentials := mopts.Credential{
                        Username: c.GetString(dconfig.SettingDbUsername),
                }
                password := c.GetString(dconfig.SettingDbPassword)
                if password != "" {
                        credentials.Password = password
                        credentials.PasswordSet = true
                }
                clientOptions.SetAuth(credentials)
        }

        if c.GetBool(dconfig.SettingDbSSL) {
                tlsConfig := &tls.Config{}
                tlsConfig.InsecureSkipVerify = c.GetBool(dconfig.SettingDbSSLSkipVerify)
                clientOptions.SetTLSConfig(tlsConfig)
        }

        // Set 10s timeout
        ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
        defer cancel()
        client, err := mongo.Connect(ctx, clientOptions)
        if err != nil {
                return nil, errors.Wrap(err, "Failed to connect to mongo server")
        }

        // Validate connection
        if err = client.Ping(ctx, nil); err != nil {
                return nil, errors.Wrap(err, "Error reaching mongo server")
        }

        mongoClient := Client{Client: *client}
        return &mongoClient, nil
}

// DataStoreMongo is the data storage service
type DataStoreMongo struct {
        // client holds the reference to the client used to communicate with the
        // mongodb server.
        client *Client
        // dbName contains the name of the workflow database.
        dbName string
        // workflows holds a local cache of workflows - a worker should NEVER
        // access this cache directly, but through
        // DataStoreMongo.GetWorkflowByName.
        workflows map[string]*model.Workflow
}

// NewDataStoreWithClient initializes a DataStore object
func NewDataStoreWithClient(client *Client, c config.Reader) *DataStoreMongo {
        dbName := c.GetString(dconfig.SettingDbName)
        ctx := context.Background()

        // Maybe initialize workflows
        var findResults []*model.Workflow
        workflows := make(map[string]*model.Workflow)
        database := client.Database(dbName)
        collWflows := database.Collection(WorkflowCollectionName)
        cur, err := collWflows.Find(ctx, bson.M{})
        if err == nil {
                if err = cur.All(ctx, &findResults); err == nil {
                        for _, workflow := range findResults {
                                workflows[workflow.Name+"."+strconv.Itoa(workflow.Version)] = workflow
                        }
                }
        }

        return &DataStoreMongo{
                client:    client,
                dbName:    dbName,
                workflows: workflows,
        }
}

func (db *DataStoreMongo) Ping(ctx context.Context) error {
        res := db.client.Database(db.dbName).
                RunCommand(ctx, bson.M{"ping": 1})
        return res.Err()
}

// LoadWorkflows from filesystem if the workflowsPath setting is provided
func (db *DataStoreMongo) LoadWorkflows(ctx context.Context, l *log.Logger) error {
        workflowsPath := config.Config.GetString(dconfig.SettingWorkflowsPath)
        if workflowsPath != "" {
                workflows := model.GetWorkflowsFromPath(workflowsPath)
                l.Infof("LoadWorkflows: loading %d workflows from %s.", len(workflows), workflowsPath)
                for _, workflow := range workflows {
                        l.Infof("LoadWorkflows: loading %s v%d.", workflow.Name, workflow.Version)
                        count, err := db.InsertWorkflows(ctx, *workflow)
                        if count != 1 {
                                l.Infof("LoadWorkflows: not loaded: %s v%d.", workflow.Name, workflow.Version)
                        }
                        if err != nil {
                                l.Infof(
                                        "LoadWorkflows: error loading: %s v%d: %s.",
                                        workflow.Name,
                                        workflow.Version,
                                        err.Error(),
                                )
                        }
                }
        } else {
                l.Info("LoadWorkflows: empty workflowsPath, not loading workflows")
        }
        return nil
}

// InsertWorkflows inserts a workflow to the database and cache and returns the number of
// inserted elements or an error for the first error generated.
func (db *DataStoreMongo) InsertWorkflows(
        ctx context.Context,
        workflows ...model.Workflow,
) (int, error) {
        database := db.client.Database(db.dbName)
        collWflows := database.Collection(WorkflowCollectionName)
        for i, workflow := range workflows {
                if workflow.Name == "" {
                        return i, store.ErrWorkflowMissingName
                }
                workflowDb, _ := db.GetWorkflowByName(ctx, workflow.Name, strconv.Itoa(workflow.Version))
                if workflowDb != nil && workflowDb.Version >= workflow.Version {
                        return i + 1, store.ErrWorkflowAlreadyExists
                }
                if workflowDb == nil || workflowDb.Version < workflow.Version {
                        upsert := true
                        opt := &mopts.UpdateOptions{
                                Upsert: &upsert,
                        }
                        query := bson.M{"_id": workflow.Name}
                        update := bson.M{"$set": workflow}
                        if _, err := collWflows.UpdateOne(ctx, query, update, opt); err != nil {
                                return i + 1, err
                        }
                }
                db.workflows[workflow.Name] = &workflow
        }
        return len(workflows), nil
}

// GetWorkflowByName gets the workflow with the given name - either from the
// cache, or searches the database if the workflow is not cached.
func (db *DataStoreMongo) GetWorkflowByName(
        ctx context.Context,
        workflowName string,
        version string,
) (*model.Workflow, error) {
        workflow, ok := db.workflows[workflowName]
        l := log.FromContext(ctx)

        versionNumber, e := strconv.Atoi(version)
        if e != nil {
                versionNumber = 0
        }

        if !ok || workflow.Version < versionNumber {
                var result model.Workflow
                database := db.client.Database(db.dbName)
                collWflows := database.Collection(WorkflowCollectionName)
                err := collWflows.FindOne(ctx, bson.M{"_id": workflowName}).
                        Decode(&result)
                if err != nil {
                        return nil, err
                }
                if result.Version < versionNumber {
                        l.Errorf("workflow found but with version too low: %s v%s", workflowName, version)
                        return nil, ErrNoSuchWorkflowByVersion
                }
                db.workflows[result.Name] = &result
                return &result, err
        } else {
                l.Debugf("cache hit: %s v%s", workflowName, version)
        }
        return workflow, nil
}

// GetWorkflows gets all workflows from the cache as a list
// (should only be used by the server process)
func (db *DataStoreMongo) GetWorkflows(ctx context.Context) []model.Workflow {
        workflows := make([]model.Workflow, len(db.workflows))
        var i int
        for _, workflow := range db.workflows {
                workflows[i] = *workflow
                i++
        }

        return workflows
}

// UpsertJob inserts the job in the queue
func (db *DataStoreMongo) UpsertJob(
        ctx context.Context, job *model.Job) (*model.Job, error) {
        if job.ID == "" {
                job.ID = primitive.NewObjectID().Hex()
        }
        query := bson.M{
                "_id": job.ID,
        }
        update := bson.M{
                "$set": job,
        }
        findUpdateOptions := &mopts.FindOneAndUpdateOptions{}
        findUpdateOptions.SetReturnDocument(mopts.After)
        findUpdateOptions.SetUpsert(true)

        database := db.client.Database(db.dbName)
        collJobs := database.Collection(JobsCollectionName)

        err := collJobs.FindOneAndUpdate(ctx, query, update, findUpdateOptions).Decode(job)
        if err != nil {
                return nil, err
        }

        return job, nil
}

// UpdateJobAddResult add a task execution result to a job status
func (db *DataStoreMongo) UpdateJobAddResult(ctx context.Context,
        job *model.Job, result *model.TaskResult) error {
        options := &mopts.UpdateOptions{}
        options.SetUpsert(true)

        update := bson.M{
                "$addToSet": bson.M{
                        "results": result,
                },
                "$setOnInsert": bson.M{
                        "workflow_name":    job.WorkflowName,
                        "input_parameters": job.InputParameters,
                        "status":           job.Status,
                        "insert_time":      job.InsertTime,
                        "version":          job.WorkflowVersion,
                },
        }

        collection := db.client.Database(db.dbName).
                Collection(JobsCollectionName)
        _, err := collection.UpdateOne(ctx, bson.M{"_id": job.ID}, update, options)
        if err != nil {
                return err
        }

        return nil
}

// UpdateJobStatus set the task execution status for a job status
func (db *DataStoreMongo) UpdateJobStatus(
        ctx context.Context, job *model.Job, status int32) error {
        if model.StatusToString(status) == "unknown" {
                return model.ErrInvalidStatus
        }

        options := &mopts.UpdateOptions{}
        options.SetUpsert(true)

        collection := db.client.Database(db.dbName).
                Collection(JobsCollectionName)
        _, err := collection.UpdateOne(ctx, bson.M{
                "_id": job.ID,
        }, bson.M{
                "$set": bson.M{
                        "status": status,
                },
                "$setOnInsert": bson.M{
                        "workflow_name":    job.WorkflowName,
                        "input_parameters": job.InputParameters,
                        "results":          job.Results,
                        "insert_time":      job.InsertTime,
                        "version":          job.WorkflowVersion,
                },
        }, options)
        if err != nil {
                return err
        }

        return nil
}

// GetJobByNameAndID get the task execution status for a job
// by workflow name and ID
func (db *DataStoreMongo) GetJobByNameAndID(
        ctx context.Context, name string, ID string) (*model.Job, error) {
        collection := db.client.Database(db.dbName).
                Collection(JobsCollectionName)
        cur := collection.FindOne(ctx, bson.M{
                "_id":           ID,
                "workflow_name": name,
        })
        var job model.Job
        err := cur.Decode(&job)
        if err == mongo.ErrNoDocuments {
                return nil, nil
        } else if err != nil {
                return nil, err
        }

        return &job, nil
}

func (db *DataStoreMongo) GetAllJobs(
        ctx context.Context, page int64, perPage int64) ([]model.Job, int64, error) {
        collection := db.client.Database(db.dbName).
                Collection(JobsCollectionName)
        findOptions := &options.FindOptions{}
        findOptions.SetSkip(int64((page - 1) * perPage))
        findOptions.SetLimit(int64(perPage))
        sortField := bson.M{}
        sortField["insert_time"] = -1
        findOptions.SetSort(sortField)
        cur, err := collection.Find(ctx, bson.M{}, findOptions)
        if err != nil {
                return []model.Job{}, 0, err
        }

        var jobs []model.Job
        err = cur.All(ctx, &jobs)
        if err == mongo.ErrNoDocuments {
                return []model.Job{}, 0, nil
        } else if err != nil {
                return []model.Job{}, 0, err
        }

        count, err := collection.CountDocuments(ctx, bson.M{})
        if err != nil {
                return []model.Job{}, 0, err
        }
        return jobs, count, nil
}

// Close disconnects the client
func (db *DataStoreMongo) Close() {
        ctx := context.Background()
        disconnectClient(ctx, db.client)
}

1	// Copyright 2022 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 mongo
16
17	import (
18	"context"
19	"crypto/tls"
20	"fmt"
21	"strconv"
22	"strings"
23	"time"
24
25	"github.com/pkg/errors"
26	"go.mongodb.org/mongo-driver/bson"
27	"go.mongodb.org/mongo-driver/bson/primitive"
28	"go.mongodb.org/mongo-driver/mongo"
29	"go.mongodb.org/mongo-driver/mongo/options"
30	mopts "go.mongodb.org/mongo-driver/mongo/options"
31
32	"github.com/mendersoftware/go-lib-micro/config"
33	"github.com/mendersoftware/go-lib-micro/log"
34
35	dconfig "github.com/mendersoftware/workflows/config"
36	"github.com/mendersoftware/workflows/model"
37	"github.com/mendersoftware/workflows/store"
38	)
39
40	const (
41	// JobsCollectionName refers to the collection of finished or
42	// jobs in progress.
43	JobsCollectionName = "jobs"
44
45	// WorkflowCollectionName refers to the collection of stored workflows
46	WorkflowCollectionName = "workflows"
47	)
48
49	var (
50	ErrNoSuchWorkflowByVersion = errors.New("Workflow of the given version not found")
51	)
52
53	// SetupDataStore returns the mongo data store and optionally runs migrations
54	func SetupDataStore(automigrate bool) (*DataStoreMongo, error) {	4✔
55	ctx := context.Background()	4✔
56	dbClient, err := NewClient(ctx, config.Config)	4✔
57	if err != nil {	4✔
58	return nil, errors.New(fmt.Sprintf("failed to connect to db: %v", err))	×
59	}	×
60	err = doMigrations(ctx, dbClient, automigrate)	4✔
61	if err != nil {	4✔
62	return nil, err	×
63	}	×
64	dataStore := NewDataStoreWithClient(dbClient, config.Config)	4✔
65	return dataStore, nil	4✔
66	}
67
68	func doMigrations(ctx context.Context, client *Client,
69	automigrate bool) error {	4✔
70	db := config.Config.GetString(dconfig.SettingDbName)	4✔
71	err := Migrate(ctx, db, DbVersion, client, automigrate)	4✔
72	if err != nil {	4✔
73	return errors.New(fmt.Sprintf("failed to run migrations: %v", err))	×
74	}	×
75
76	return nil	4✔
77	}
78
79	func disconnectClient(parentCtx context.Context, client *Client) {	2✔
80	ctx, cancel := context.WithTimeout(parentCtx, 10*time.Second)	2✔
81	defer cancel()	2✔
82	_ = client.Disconnect(ctx)	2✔
83	}	2✔
84
85	// Client is a package specific mongo client
86	type Client struct {
87	mongo.Client
88	}
89
90	// NewClient returns a mongo client
91	func NewClient(_ context.Context, c config.Reader) (*Client, error) {	4✔
92		4✔
93	clientOptions := mopts.Client()	4✔
94	mongoURL := c.GetString(dconfig.SettingMongo)	4✔
95	if !strings.Contains(mongoURL, "://") {	4✔
96	return nil, errors.Errorf("Invalid mongoURL %q: missing schema.",	×
97	mongoURL)	×
98	}	×
99	clientOptions.ApplyURI(mongoURL)	4✔
100		4✔
101	username := c.GetString(dconfig.SettingDbUsername)	4✔
102	if username != "" {	4✔
103	credentials := mopts.Credential{	×
104	Username: c.GetString(dconfig.SettingDbUsername),	×
105	}	×
106	password := c.GetString(dconfig.SettingDbPassword)	×
107	if password != "" {	×
108	credentials.Password = password	×
109	credentials.PasswordSet = true	×
110	}	×
111	clientOptions.SetAuth(credentials)	×
112	}
113
114	if c.GetBool(dconfig.SettingDbSSL) {	4✔
115	tlsConfig := &tls.Config{}	×
116	tlsConfig.InsecureSkipVerify = c.GetBool(dconfig.SettingDbSSLSkipVerify)	×
117	clientOptions.SetTLSConfig(tlsConfig)	×
118	}	×
119
120	// Set 10s timeout
121	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)	4✔
122	defer cancel()	4✔
123	client, err := mongo.Connect(ctx, clientOptions)	4✔
124	if err != nil {	4✔
125	return nil, errors.Wrap(err, "Failed to connect to mongo server")	×
126	}	×
127
128	// Validate connection
129	if err = client.Ping(ctx, nil); err != nil {	4✔
130	return nil, errors.Wrap(err, "Error reaching mongo server")	×
131	}	×
132
133	mongoClient := Client{Client: *client}	4✔
134	return &mongoClient, nil	4✔
135	}
136
137	// DataStoreMongo is the data storage service
138	type DataStoreMongo struct {
139	// client holds the reference to the client used to communicate with the
140	// mongodb server.
141	client *Client
142	// dbName contains the name of the workflow database.
143	dbName string
144	// workflows holds a local cache of workflows - a worker should NEVER
145	// access this cache directly, but through
146	// DataStoreMongo.GetWorkflowByName.
147	workflows map[string]*model.Workflow
148	}
149
150	// NewDataStoreWithClient initializes a DataStore object
151	func NewDataStoreWithClient(client Client, c config.Reader) DataStoreMongo {	4✔
152	dbName := c.GetString(dconfig.SettingDbName)	4✔
153	ctx := context.Background()	4✔
154		4✔
155	// Maybe initialize workflows	4✔
156	var findResults []*model.Workflow	4✔
157	workflows := make(map[string]*model.Workflow)	4✔
158	database := client.Database(dbName)	4✔
159	collWflows := database.Collection(WorkflowCollectionName)	4✔
160	cur, err := collWflows.Find(ctx, bson.M{})	4✔
161	if err == nil {	8✔
162	if err = cur.All(ctx, &findResults); err == nil {	8✔
163	for _, workflow := range findResults {	4✔
164	workflows[workflow.Name+"."+strconv.Itoa(workflow.Version)] = workflow	×
165	}	×
166	}
167	}
168
169	return &DataStoreMongo{	4✔
170	client: client,	4✔
171	dbName: dbName,	4✔
172	workflows: workflows,	4✔
173	}	4✔
174	}
175
176	func (db *DataStoreMongo) Ping(ctx context.Context) error {	2✔
177	res := db.client.Database(db.dbName).	2✔
178	RunCommand(ctx, bson.M{"ping": 1})	2✔
179	return res.Err()	2✔
180	}	2✔
181
182	// LoadWorkflows from filesystem if the workflowsPath setting is provided
183	func (db DataStoreMongo) LoadWorkflows(ctx context.Context, l log.Logger) error {	4✔
184	workflowsPath := config.Config.GetString(dconfig.SettingWorkflowsPath)	4✔
185	if workflowsPath != "" {	7✔
186	workflows := model.GetWorkflowsFromPath(workflowsPath)	3✔
187	l.Infof("LoadWorkflows: loading %d workflows from %s.", len(workflows), workflowsPath)	3✔
188	for _, workflow := range workflows {	15✔
189	l.Infof("LoadWorkflows: loading %s v%d.", workflow.Name, workflow.Version)	12✔
190	count, err := db.InsertWorkflows(ctx, *workflow)	12✔
191	if count != 1 {	13✔
192	l.Infof("LoadWorkflows: not loaded: %s v%d.", workflow.Name, workflow.Version)	1✔
193	}	1✔
194	if err != nil {	13✔
195	l.Infof(	1✔
196	"LoadWorkflows: error loading: %s v%d: %s.",	1✔
197	workflow.Name,	1✔
198	workflow.Version,	1✔
199	err.Error(),	1✔
200	)	1✔
201	}	1✔
202	}
203	} else {	1✔
204	l.Info("LoadWorkflows: empty workflowsPath, not loading workflows")	1✔
205	}	1✔
206	return nil	4✔
207	}
208
209	// InsertWorkflows inserts a workflow to the database and cache and returns the number of
210	// inserted elements or an error for the first error generated.
211	func (db *DataStoreMongo) InsertWorkflows(
212	ctx context.Context,
213	workflows ...model.Workflow,
214	) (int, error) {	30✔
215	database := db.client.Database(db.dbName)	30✔
216	collWflows := database.Collection(WorkflowCollectionName)	30✔
217	for i, workflow := range workflows {	62✔
218	if workflow.Name == "" {	35✔
219	return i, store.ErrWorkflowMissingName	3✔
220	}	3✔
221	workflowDb, _ := db.GetWorkflowByName(ctx, workflow.Name, strconv.Itoa(workflow.Version))	29✔
222	if workflowDb != nil && workflowDb.Version >= workflow.Version {	35✔
223	return i + 1, store.ErrWorkflowAlreadyExists	6✔
224	}	6✔
225	if workflowDb == nil \|\| workflowDb.Version < workflow.Version {	46✔
226	upsert := true	23✔
227	opt := &mopts.UpdateOptions{	23✔
228	Upsert: &upsert,	23✔
229	}	23✔
230	query := bson.M{"_id": workflow.Name}	23✔
231	update := bson.M{"$set": workflow}	23✔
232	if _, err := collWflows.UpdateOne(ctx, query, update, opt); err != nil {	23✔
233	return i + 1, err	×
234	}	×
235	}
236	db.workflows[workflow.Name] = &workflow	23✔
237	}
238	return len(workflows), nil	21✔
239	}
240
241	// GetWorkflowByName gets the workflow with the given name - either from the
242	// cache, or searches the database if the workflow is not cached.
243	func (db *DataStoreMongo) GetWorkflowByName(
244	ctx context.Context,
245	workflowName string,
246	version string,
247	) (*model.Workflow, error) {	64✔
248	workflow, ok := db.workflows[workflowName]	64✔
249	l := log.FromContext(ctx)	64✔
250		64✔
251	versionNumber, e := strconv.Atoi(version)	64✔
252	if e != nil {	85✔
253	versionNumber = 0	21✔
254	}	21✔
255
256	if !ok \|\| workflow.Version < versionNumber {	102✔
257	var result model.Workflow	38✔
258	database := db.client.Database(db.dbName)	38✔
259	collWflows := database.Collection(WorkflowCollectionName)	38✔
260	err := collWflows.FindOne(ctx, bson.M{"_id": workflowName}).	38✔
261	Decode(&result)	38✔
262	if err != nil {	66✔
263	return nil, err	28✔
264	}	28✔
265	if result.Version < versionNumber {	10✔
266	l.Errorf("workflow found but with version too low: %s v%s", workflowName, version)	×
267	return nil, ErrNoSuchWorkflowByVersion	×
268	}	×
269	db.workflows[result.Name] = &result	10✔
270	return &result, err	10✔
271	} else {	26✔
272	l.Debugf("cache hit: %s v%s", workflowName, version)	26✔
273	}	26✔
274	return workflow, nil	26✔
275	}
276
277	// GetWorkflows gets all workflows from the cache as a list
278	// (should only be used by the server process)
279	func (db *DataStoreMongo) GetWorkflows(ctx context.Context) []model.Workflow {	2✔
280	workflows := make([]model.Workflow, len(db.workflows))	2✔
281	var i int	2✔
282	for _, workflow := range db.workflows {	4✔
283	workflows[i] = *workflow	2✔
284	i++	2✔
285	}	2✔
286
287	return workflows	2✔
288	}
289
290	// UpsertJob inserts the job in the queue
291	func (db *DataStoreMongo) UpsertJob(
292	ctx context.Context, job model.Job) (model.Job, error) {	20✔
293	if job.ID == "" {	30✔
294	job.ID = primitive.NewObjectID().Hex()	10✔
295	}	10✔
296	query := bson.M{	20✔
297	"_id": job.ID,	20✔
298	}	20✔
299	update := bson.M{	20✔
300	"$set": job,	20✔
301	}	20✔
302	findUpdateOptions := &mopts.FindOneAndUpdateOptions{}	20✔
303	findUpdateOptions.SetReturnDocument(mopts.After)	20✔
304	findUpdateOptions.SetUpsert(true)	20✔
305		20✔
306	database := db.client.Database(db.dbName)	20✔
307	collJobs := database.Collection(JobsCollectionName)	20✔
308		20✔
309	err := collJobs.FindOneAndUpdate(ctx, query, update, findUpdateOptions).Decode(job)	20✔
310	if err != nil {	20✔
311	return nil, err	×
312	}	×
313
314	return job, nil	20✔
315	}
316
317	// UpdateJobAddResult add a task execution result to a job status
318	func (db *DataStoreMongo) UpdateJobAddResult(ctx context.Context,
319	job model.Job, result model.TaskResult) error {	30✔
320	options := &mopts.UpdateOptions{}	30✔
321	options.SetUpsert(true)	30✔
322		30✔
323	update := bson.M{	30✔
324	"$addToSet": bson.M{	30✔
325	"results": result,	30✔
326	},	30✔
327	"$setOnInsert": bson.M{	30✔
328	"workflow_name": job.WorkflowName,	30✔
329	"input_parameters": job.InputParameters,	30✔
330	"status": job.Status,	30✔
331	"insert_time": job.InsertTime,	30✔
332	"version": job.WorkflowVersion,	30✔
333	},	30✔
334	}	30✔
335		30✔
336	collection := db.client.Database(db.dbName).	30✔
337	Collection(JobsCollectionName)	30✔
338	_, err := collection.UpdateOne(ctx, bson.M{"_id": job.ID}, update, options)	30✔
339	if err != nil {	30✔
340	return err	×
341	}	×
342
343	return nil	30✔
344	}
345
346	// UpdateJobStatus set the task execution status for a job status
347	func (db *DataStoreMongo) UpdateJobStatus(
348	ctx context.Context, job *model.Job, status int32) error {	14✔
349	if model.StatusToString(status) == "unknown" {	16✔
350	return model.ErrInvalidStatus	2✔
351	}	2✔
352
353	options := &mopts.UpdateOptions{}	12✔
354	options.SetUpsert(true)	12✔
355		12✔
356	collection := db.client.Database(db.dbName).	12✔
357	Collection(JobsCollectionName)	12✔
358	_, err := collection.UpdateOne(ctx, bson.M{	12✔
359	"_id": job.ID,	12✔
360	}, bson.M{	12✔
361	"$set": bson.M{	12✔
362	"status": status,	12✔
363	},	12✔
364	"$setOnInsert": bson.M{	12✔
365	"workflow_name": job.WorkflowName,	12✔
366	"input_parameters": job.InputParameters,	12✔
367	"results": job.Results,	12✔
368	"insert_time": job.InsertTime,	12✔
369	"version": job.WorkflowVersion,	12✔
370	},	12✔
371	}, options)	12✔
372	if err != nil {	12✔
373	return err	×
374	}	×
375
376	return nil	12✔
377	}
378
379	// GetJobByNameAndID get the task execution status for a job
380	// by workflow name and ID
381	func (db *DataStoreMongo) GetJobByNameAndID(
382	ctx context.Context, name string, ID string) (*model.Job, error) {	34✔
383	collection := db.client.Database(db.dbName).	34✔
384	Collection(JobsCollectionName)	34✔
385	cur := collection.FindOne(ctx, bson.M{	34✔
386	"_id": ID,	34✔
387	"workflow_name": name,	34✔
388	})	34✔
389	var job model.Job	34✔
390	err := cur.Decode(&job)	34✔
391	if err == mongo.ErrNoDocuments {	34✔
392	return nil, nil	×
393	} else if err != nil {	34✔
394	return nil, err	×
395	}	×
396
397	return &job, nil	34✔
398	}
399
400	func (db *DataStoreMongo) GetAllJobs(
401	ctx context.Context, page int64, perPage int64) ([]model.Job, int64, error) {	2✔
402	collection := db.client.Database(db.dbName).	2✔
403	Collection(JobsCollectionName)	2✔
404	findOptions := &options.FindOptions{}	2✔
405	findOptions.SetSkip(int64((page - 1) * perPage))	2✔
406	findOptions.SetLimit(int64(perPage))	2✔
407	sortField := bson.M{}	2✔
408	sortField["insert_time"] = -1	2✔
409	findOptions.SetSort(sortField)	2✔
410	cur, err := collection.Find(ctx, bson.M{}, findOptions)	2✔
411	if err != nil {	2✔
412	return []model.Job{}, 0, err	×
413	}	×
414
415	var jobs []model.Job	2✔
416	err = cur.All(ctx, &jobs)	2✔
417	if err == mongo.ErrNoDocuments {	2✔
418	return []model.Job{}, 0, nil	×
419	} else if err != nil {	2✔
420	return []model.Job{}, 0, err	×
421	}	×
422
423	count, err := collection.CountDocuments(ctx, bson.M{})	2✔
424	if err != nil {	2✔
425	return []model.Job{}, 0, err	×
426	}	×
427	return jobs, count, nil	2✔
428	}
429
430	// Close disconnects the client
431	func (db *DataStoreMongo) Close() {	2✔
432	ctx := context.Background()	2✔
433	disconnectClient(ctx, db.client)	2✔
434	}	2✔

mendersoftware / workflows / 792065958

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