938326290

pending completion

Build # 938326290

Build Type

push

gitlab-ci

Committed by

web-flow

Commit Message

Merge pull request #251 from lluiscampos/chore-update-dind

ci: Update Docker-inDocker service images to latest

Run Details

2546 of 3223 relevant lines covered (78.99%)

22.83 hits per line

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

75.88

/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"
        "io"
        "strings"
        "time"

        "github.com/google/uuid"
        "github.com/pkg/errors"
        "github.com/vmihailenco/msgpack/v5"
        "go.mongodb.org/mongo-driver/bson"
        "go.mongodb.org/mongo-driver/mongo"
        mopts "go.mongodb.org/mongo-driver/mongo/options"
        "go.mongodb.org/mongo-driver/mongo/writeconcern"

        "github.com/mendersoftware/go-lib-micro/config"
        "github.com/mendersoftware/go-lib-micro/identity"
        "github.com/mendersoftware/go-lib-micro/log"
        mdoc "github.com/mendersoftware/go-lib-micro/mongo/doc"
        "github.com/mendersoftware/go-lib-micro/mongo/migrate"
        mstore "github.com/mendersoftware/go-lib-micro/store/v2"
        "github.com/mendersoftware/go-lib-micro/ws"
        "github.com/mendersoftware/go-lib-micro/ws/shell"

        "github.com/mendersoftware/deviceconnect/app"
        dconfig "github.com/mendersoftware/deviceconnect/config"
        "github.com/mendersoftware/deviceconnect/model"
        "github.com/mendersoftware/deviceconnect/store"
        "github.com/mendersoftware/deviceconnect/utils"
)

var (
        clock                        utils.Clock = utils.RealClock{}
        recordingReadBufferSize                  = 1024
        ErrUnknownControlMessageType             = errors.New("unknown control message type")
        ErrRecordingDataInconsistent             = errors.New("recording data corrupt")
)

const (
        // DevicesCollectionName refers to the name of the collection of stored devices
        DevicesCollectionName = "devices"

        // SessionsCollectionName refers to the name of the collection of sessions
        SessionsCollectionName = "sessions"

        // RecordingsCollectionName name of the collection of session recordings
        RecordingsCollectionName = "recordings"

        // ControlCollectionName name of the collection of session control data
        ControlCollectionName = "control"

        dbFieldID        = "_id"
        dbFieldSessionID = "session_id"
        dbFieldDeviceID  = "device_id"
        dbFieldStatus    = "status"
        dbFieldCreatedTs = "created_ts"
        dbFieldUpdatedTs = "updated_ts"
)

// SetupDataStore returns the mongo data store and optionally runs migrations
func SetupDataStore(automigrate bool) (store.DataStore, 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,
                time.Second*time.Duration(config.Config.GetInt(dconfig.SettingRecordingExpireSec)))
        return dataStore, nil
}

func doMigrations(ctx context.Context, client *mongo.Client,
        automigrate bool) error {
        db := config.Config.GetString(dconfig.SettingDbName)
        dbs, err := migrate.GetTenantDbs(ctx, client, mstore.IsTenantDb(db))
        if err != nil {
                return errors.Wrap(err, "failed go retrieve tenant DBs")
        }
        if len(dbs) == 0 {
                dbs = []string{DbName}
        }

        for _, d := range dbs {
                err := Migrate(ctx, d, DbVersion, client, automigrate)
                if err != nil {
                        return errors.New(fmt.Sprintf("failed to run migrations: %v", err))
                }
        }
        return nil
}

// NewClient returns a mongo client
func NewClient(ctx context.Context, c config.Reader) (*mongo.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 writeconcern to acknowlage after write has propagated to the
        // mongod instance and commited to the file system journal.
        var wc *writeconcern.WriteConcern
        wc.WithOptions(writeconcern.W(1), writeconcern.J(true))
        clientOptions.SetWriteConcern(wc)

        // Set 10s timeout
        if _, ok := ctx.Deadline(); !ok {
                var cancel context.CancelFunc
                ctx, cancel = context.WithTimeout(ctx, 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")
        }

        return client, 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          *mongo.Client
        recordingExpire time.Duration
}

// NewDataStoreWithClient initializes a DataStore object
func NewDataStoreWithClient(client *mongo.Client, expire time.Duration) store.DataStore {
        return &DataStoreMongo{
                client:          client,
                recordingExpire: expire,
        }
}

// Ping verifies the connection to the database
func (db *DataStoreMongo) Ping(ctx context.Context) error {
        res := db.client.Database(DbName).RunCommand(ctx, bson.M{"ping": 1})
        return res.Err()
}

// ProvisionDevice provisions a new device
func (db *DataStoreMongo) ProvisionDevice(ctx context.Context, tenantID, deviceID string) error {
        coll := db.client.Database(DbName).Collection(DevicesCollectionName)

        now := clock.Now().UTC()

        updateOpts := &mopts.UpdateOptions{}
        updateOpts.SetUpsert(true)
        _, err := coll.UpdateOne(ctx,
                bson.M{dbFieldID: deviceID, mstore.FieldTenantID: tenantID},
                bson.M{
                        "$setOnInsert": bson.M{
                                dbFieldStatus:        model.DeviceStatusUnknown,
                                dbFieldCreatedTs:     &now,
                                dbFieldUpdatedTs:     &now,
                                mstore.FieldTenantID: tenantID,
                        },
                },
                updateOpts,
        )
        return err
}

// DeleteDevice deletes a device
func (db *DataStoreMongo) DeleteDevice(ctx context.Context, tenantID, deviceID string) error {
        coll := db.client.Database(DbName).Collection(DevicesCollectionName)

        _, err := coll.DeleteOne(ctx, bson.M{dbFieldID: deviceID, mstore.FieldTenantID: tenantID})
        return err
}

// GetDevice returns a device
func (db *DataStoreMongo) GetDevice(
        ctx context.Context,
        tenantID string,
        deviceID string,
) (*model.Device, error) {
        coll := db.client.Database(DbName).Collection(DevicesCollectionName)

        cur := coll.FindOne(ctx, bson.M{dbFieldID: deviceID, mstore.FieldTenantID: tenantID})

        device := &model.Device{}
        if err := cur.Decode(&device); err != nil {
                if err == mongo.ErrNoDocuments {
                        return nil, nil
                }
                return nil, err
        }

        return device, nil
}

// UpsertDeviceStatus upserts the connection status of a device
func (db *DataStoreMongo) UpsertDeviceStatus(
        ctx context.Context,
        tenantID string,
        deviceID string,
        status string,
) error {
        coll := db.client.Database(DbName).Collection(DevicesCollectionName)

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

        now := clock.Now().UTC()

        _, err := coll.UpdateOne(ctx,
                bson.M{dbFieldID: deviceID, mstore.FieldTenantID: tenantID},
                bson.M{
                        "$set": bson.M{
                                dbFieldStatus:    status,
                                dbFieldUpdatedTs: &now,
                        },
                        "$setOnInsert": bson.M{
                                dbFieldCreatedTs:     &now,
                                mstore.FieldTenantID: tenantID,
                        },
                },
                updateOpts,
        )

        return err
}

// AllocateSession allocates a new session.
func (db *DataStoreMongo) AllocateSession(ctx context.Context, sess *model.Session) error {

        if err := sess.Validate(); err != nil {
                return errors.Wrap(err, "store: cannot allocate invalid Session")
        }

        coll := db.client.Database(DbName).Collection(SessionsCollectionName)
        tenantElem := bson.E{Key: mstore.FieldTenantID, Value: sess.TenantID}
        _, err := coll.InsertOne(ctx, mdoc.DocumentFromStruct(*sess, tenantElem))
        if err != nil {
                return errors.Wrap(err, "store: failed to allocate session")
        }

        return nil
}

// DeleteSession deletes a session
func (db *DataStoreMongo) DeleteSession(
        ctx context.Context, sessionID string,
) (*model.Session, error) {
        collSess := db.client.Database(DbName).
                Collection(SessionsCollectionName)

        sess := new(model.Session)
        err := collSess.FindOneAndDelete(
                ctx, mstore.WithTenantID(ctx, bson.D{{Key: dbFieldID, Value: sessionID}}),
        ).Decode(sess)
        if err != nil {
                if err == mongo.ErrNoDocuments {
                        return nil, store.ErrSessionNotFound
                } else {
                        return nil, err
                }
        }
        if idty := identity.FromContext(ctx); idty != nil {
                sess.TenantID = idty.Tenant
        }
        return sess, nil
}

// GetSession returns a session
func (db *DataStoreMongo) GetSession(
        ctx context.Context,
        sessionID string,
) (*model.Session, error) {
        collSess := db.client.
                Database(DbName).
                Collection(SessionsCollectionName)

        session := &model.Session{}
        err := collSess.
                FindOne(ctx, mstore.WithTenantID(ctx, bson.M{dbFieldID: sessionID})).
                Decode(session)
        if err != nil {
                if err == mongo.ErrNoDocuments {
                        return nil, store.ErrSessionNotFound
                }
                return nil, err
        }
        idty := identity.FromContext(ctx)
        if idty != nil {
                session.TenantID = idty.Tenant
        }

        return session, nil
}

func sendControlMessage(control app.Control, sessionID string, w io.Writer) (int, error) {
        messageType := ""
        var data []byte
        properties := make(map[string]interface{})

        switch control.Type {
        case app.DelayMessage:
                messageType = model.DelayMessageName
                properties[model.DelayMessageValueField] = control.DelayMs
        case app.ResizeMessage:
                messageType = shell.MessageTypeResizeShell
                properties[model.ResizeMessageTermHeightField] = control.TerminalHeight
                properties[model.ResizeMessageTermWidthField] = control.TerminalWidth
        default:
                return 0, ErrUnknownControlMessageType
        }

        msg := ws.ProtoMsg{
                Header: ws.ProtoHdr{
                        Proto:      ws.ProtoTypeShell,
                        MsgType:    messageType,
                        SessionID:  sessionID,
                        Properties: properties,
                },
                Body: data,
        }
        messagePacked, err := msgpack.Marshal(&msg)
        if err != nil {
                return 0, err
        } else {
                return w.Write(messagePacked)
        }
}

// GetSession writes session recordings to given io.Writer
func (db *DataStoreMongo) WriteSessionRecords(ctx context.Context,
        sessionID string,
        w io.Writer) error {
        l := log.FromContext(ctx)
        collRecording := db.client.Database(DbName).
                Collection(RecordingsCollectionName)
        collControl := db.client.Database(DbName).
                Collection(ControlCollectionName)

        findOptions := mopts.Find()
        sortField := bson.M{
                "created_ts": 1,
        }
        findOptions.SetSort(sortField)
        recordingsCursor, err := collRecording.Find(ctx,
                mstore.WithTenantID(ctx, bson.M{
                        dbFieldSessionID: sessionID,
                }),
                findOptions,
        )
        if err != nil {
                return err
        }
        defer recordingsCursor.Close(ctx)

        controlCursor, err := collControl.Find(ctx,
                mstore.WithTenantID(ctx, bson.M{
                        dbFieldSessionID: sessionID,
                }),
                findOptions,
        )
        if err != nil {
                return err
        }
        defer controlCursor.Close(ctx)

        controlReader := NewControlMessageReader(ctx, controlCursor)
        recordingReader := NewRecordingReader(ctx, recordingsCursor)

        recordingWriter := NewRecordingWriter(sessionID, w)
        recordingBuffer := make([]byte, recordingReadBufferSize)
        recordingBytesSent := 0
        for {
                control := controlReader.Pop()
                if control == nil {
                        l.Debug("WriteSessionRecords: no more control " +
                                "messages, flushing the recording upstream.")
                        //no more control messages, we send the whole recording
                        n, err := io.Copy(recordingWriter, recordingReader)
                        if err != nil && err != io.ErrShortWrite && n < 1 {
                                l.Errorf("WriteSessionRecords: "+
                                        "error writing recording data, err: %+v n:%d",
                                        err, n)
                        }
                        if n == 0 {
                                l.Errorf("WriteSessionRecords: "+
                                        "failed to write any recording data, err: %+v",
                                        err)
                        } else {
                                recordingBytesSent += int(n)
                        }
                        break
                } else {
                        if recordingBytesSent > control.Offset {
                                //this should never happen, we missed
                                //the control message, data inconsistency
                                l.Errorf("WriteSessionRecords: recordingBytesSent > control.Offset")
                                err = ErrRecordingDataInconsistent
                                break
                        }

                        bytesUntilControlMessage := control.Offset - recordingBytesSent
                        l.Debugf("(1) WriteSessionRecords: control.Offset:%d"+
                                " recordingBytesSent:%d "+
                                " bytesUntilControlMessage:%d (recordingBuffer.len=%d) "+
                                "reading up to %d bytes of recording and sending.",
                                control.Offset,
                                recordingBytesSent,
                                bytesUntilControlMessage,
                                len(recordingBuffer),
                                control.Offset-recordingBytesSent)
                        //it is possible that the recording is larger than one recordingBuffer,
                        //we need to send until we have the control.Offset in the buffer
                        for bytesUntilControlMessage > len(recordingBuffer) {
                                n, e := recordingReader.Read(recordingBuffer)
                                if n > 0 {
                                        _, err = sendRecordingMessage(recordingBuffer[:n],
                                                sessionID,
                                                w)
                                        if err != nil {
                                                l.Errorf("error sending recording data: %s",
                                                        err.Error())
                                                break
                                        }
                                        recordingBytesSent += n
                                        bytesUntilControlMessage = control.Offset -
                                                recordingBytesSent
                                }
                                if e != nil || n == 0 {
                                        break
                                }
                        }
                        if err != nil {
                                break
                        }

                        bytesUntilControlMessage = control.Offset - recordingBytesSent
                        l.Debugf("(2) WriteSessionRecords: control.Offset:%d"+
                                " recordingBytesSent:%d "+
                                " bytesUntilControlMessage:%d (recordingBuffer.len=%d) "+
                                "reading up to %d bytes of recording and sending.",
                                control.Offset,
                                recordingBytesSent,
                                bytesUntilControlMessage,
                                len(recordingBuffer),
                                bytesUntilControlMessage)
                        //this means that the control offset is in the future
                        //part of the recording buffer
                        //we can send up to control.Offset-recordingBytesSent
                        //bytes and then send the control message
                        n, e := recordingReader.Read(recordingBuffer[:bytesUntilControlMessage])
                        l.Debugf("recordingReader.Read(len=%d)=%d,%+v",
                                bytesUntilControlMessage, n, e)
                        if n > 0 {
                                _, err = sendRecordingMessage(recordingBuffer[:n], sessionID, w)
                                if err != nil {
                                        l.Errorf("error sending recording data: %s",
                                                err.Error())
                                        break
                                }
                                recordingBytesSent += n
                        }
                        l.Debugf("WriteSessionRecords: sending %+v.", *control)
                        _, err = sendControlMessage(*control, sessionID, w)
                        if err != nil {
                                l.Errorf("error sending recording data: %s",
                                        err.Error())
                                break
                        }
                }
        }
        l.Infof("session playback: WriteSessionRecords: sent %d bytes.", recordingBytesSent)

        return err
}

// SetSession saves a session recording
func (db *DataStoreMongo) InsertSessionRecording(ctx context.Context,
        sessionID string,
        sessionBytes []byte) error {
        coll := db.client.Database(DbName).Collection(RecordingsCollectionName)

        now := clock.Now().UTC()
        recording := model.Recording{
                ID:        uuid.New(),
                SessionID: sessionID,
                Recording: sessionBytes,
                CreatedTs: now,
                ExpireTs:  now.Add(db.recordingExpire),
        }
        _, err := coll.InsertOne(ctx,
                mstore.WithTenantID(ctx, &recording),
        )
        return err
}

// Inserts control data recording
func (db *DataStoreMongo) InsertControlRecording(ctx context.Context,
        sessionID string,
        sessionBytes []byte) error {
        coll := db.client.Database(DbName).
                Collection(ControlCollectionName)

        now := clock.Now().UTC()
        recording := model.ControlData{
                ID:        uuid.New(),
                SessionID: sessionID,
                Control:   sessionBytes,
                CreatedTs: now,
                ExpireTs:  now.Add(db.recordingExpire),
        }
        _, err := coll.InsertOne(ctx,
                mstore.WithTenantID(ctx, &recording),
        )
        return err
}

// Close disconnects the client
func (db *DataStoreMongo) Close() error {
        ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
        defer cancel()
        err := db.client.Disconnect(ctx)
        return err
}

//nolint:unused
func (db *DataStoreMongo) DropDatabase() error {
        ctx := context.Background()
        err := db.client.Database(DbName).Drop(ctx)
        return err
}

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	"io"
22	"strings"
23	"time"
24
25	"github.com/google/uuid"
26	"github.com/pkg/errors"
27	"github.com/vmihailenco/msgpack/v5"
28	"go.mongodb.org/mongo-driver/bson"
29	"go.mongodb.org/mongo-driver/mongo"
30	mopts "go.mongodb.org/mongo-driver/mongo/options"
31	"go.mongodb.org/mongo-driver/mongo/writeconcern"
32
33	"github.com/mendersoftware/go-lib-micro/config"
34	"github.com/mendersoftware/go-lib-micro/identity"
35	"github.com/mendersoftware/go-lib-micro/log"
36	mdoc "github.com/mendersoftware/go-lib-micro/mongo/doc"
37	"github.com/mendersoftware/go-lib-micro/mongo/migrate"
38	mstore "github.com/mendersoftware/go-lib-micro/store/v2"
39	"github.com/mendersoftware/go-lib-micro/ws"
40	"github.com/mendersoftware/go-lib-micro/ws/shell"
41
42	"github.com/mendersoftware/deviceconnect/app"
43	dconfig "github.com/mendersoftware/deviceconnect/config"
44	"github.com/mendersoftware/deviceconnect/model"
45	"github.com/mendersoftware/deviceconnect/store"
46	"github.com/mendersoftware/deviceconnect/utils"
47	)
48
49	var (
50	clock utils.Clock = utils.RealClock{}
51	recordingReadBufferSize = 1024
52	ErrUnknownControlMessageType = errors.New("unknown control message type")
53	ErrRecordingDataInconsistent = errors.New("recording data corrupt")
54	)
55
56	const (
57	// DevicesCollectionName refers to the name of the collection of stored devices
58	DevicesCollectionName = "devices"
59
60	// SessionsCollectionName refers to the name of the collection of sessions
61	SessionsCollectionName = "sessions"
62
63	// RecordingsCollectionName name of the collection of session recordings
64	RecordingsCollectionName = "recordings"
65
66	// ControlCollectionName name of the collection of session control data
67	ControlCollectionName = "control"
68
69	dbFieldID = "_id"
70	dbFieldSessionID = "session_id"
71	dbFieldDeviceID = "device_id"
72	dbFieldStatus = "status"
73	dbFieldCreatedTs = "created_ts"
74	dbFieldUpdatedTs = "updated_ts"
75	)
76
77	// SetupDataStore returns the mongo data store and optionally runs migrations
78	func SetupDataStore(automigrate bool) (store.DataStore, error) {	1✔
79	ctx := context.Background()	1✔
80	dbClient, err := NewClient(ctx, config.Config)	1✔
81	if err != nil {	1✔
82	return nil, errors.New(fmt.Sprintf("failed to connect to db: %v", err))	×
83	}	×
84	err = doMigrations(ctx, dbClient, automigrate)	1✔
85	if err != nil {	1✔
86	return nil, err	×
87	}	×
88	dataStore := NewDataStoreWithClient(dbClient,	1✔
89	time.Second*time.Duration(config.Config.GetInt(dconfig.SettingRecordingExpireSec)))	1✔
90	return dataStore, nil	1✔
91	}
92
93	func doMigrations(ctx context.Context, client *mongo.Client,
94	automigrate bool) error {	1✔
95	db := config.Config.GetString(dconfig.SettingDbName)	1✔
96	dbs, err := migrate.GetTenantDbs(ctx, client, mstore.IsTenantDb(db))	1✔
97	if err != nil {	1✔
98	return errors.Wrap(err, "failed go retrieve tenant DBs")	×
99	}	×
100	if len(dbs) == 0 {	2✔
101	dbs = []string{DbName}	1✔
102	}	1✔
103
104	for _, d := range dbs {	2✔
105	err := Migrate(ctx, d, DbVersion, client, automigrate)	1✔
106	if err != nil {	1✔
107	return errors.New(fmt.Sprintf("failed to run migrations: %v", err))	×
108	}	×
109	}
110	return nil	1✔
111	}
112
113	// NewClient returns a mongo client
114	func NewClient(ctx context.Context, c config.Reader) (*mongo.Client, error) {	1✔
115		1✔
116	clientOptions := mopts.Client()	1✔
117	mongoURL := c.GetString(dconfig.SettingMongo)	1✔
118	if !strings.Contains(mongoURL, "://") {	1✔
119	return nil, errors.Errorf("Invalid mongoURL %q: missing schema.",	×
120	mongoURL)	×
121	}	×
122	clientOptions.ApplyURI(mongoURL)	1✔
123		1✔
124	username := c.GetString(dconfig.SettingDbUsername)	1✔
125	if username != "" {	1✔
126	credentials := mopts.Credential{	×
127	Username: c.GetString(dconfig.SettingDbUsername),	×
128	}	×
129	password := c.GetString(dconfig.SettingDbPassword)	×
130	if password != "" {	×
131	credentials.Password = password	×
132	credentials.PasswordSet = true	×
133	}	×
134	clientOptions.SetAuth(credentials)	×
135	}
136
137	if c.GetBool(dconfig.SettingDbSSL) {	1✔
138	tlsConfig := &tls.Config{}	×
139	tlsConfig.InsecureSkipVerify = c.GetBool(dconfig.SettingDbSSLSkipVerify)	×
140	clientOptions.SetTLSConfig(tlsConfig)	×
141	}	×
142
143	// Set writeconcern to acknowlage after write has propagated to the
144	// mongod instance and commited to the file system journal.
145	var wc *writeconcern.WriteConcern	1✔
146	wc.WithOptions(writeconcern.W(1), writeconcern.J(true))	1✔
147	clientOptions.SetWriteConcern(wc)	1✔
148		1✔
149	// Set 10s timeout	1✔
150	if _, ok := ctx.Deadline(); !ok {	2✔
151	var cancel context.CancelFunc	1✔
152	ctx, cancel = context.WithTimeout(ctx, 10*time.Second)	1✔
153	defer cancel()	1✔
154	}	1✔
155	client, err := mongo.Connect(ctx, clientOptions)	1✔
156	if err != nil {	1✔
157	return nil, errors.Wrap(err, "Failed to connect to mongo server")	×
158	}	×
159
160	// Validate connection
161	if err = client.Ping(ctx, nil); err != nil {	1✔
162	return nil, errors.Wrap(err, "Error reaching mongo server")	×
163	}	×
164
165	return client, nil	1✔
166	}
167
168	// DataStoreMongo is the data storage service
169	type DataStoreMongo struct {
170	// client holds the reference to the client used to communicate with the
171	// mongodb server.
172	client *mongo.Client
173	recordingExpire time.Duration
174	}
175
176	// NewDataStoreWithClient initializes a DataStore object
177	func NewDataStoreWithClient(client *mongo.Client, expire time.Duration) store.DataStore {	2✔
178	return &DataStoreMongo{	2✔
179	client: client,	2✔
180	recordingExpire: expire,	2✔
181	}	2✔
182	}	2✔
183
184	// Ping verifies the connection to the database
185	func (db *DataStoreMongo) Ping(ctx context.Context) error {	1✔
186	res := db.client.Database(DbName).RunCommand(ctx, bson.M{"ping": 1})	1✔
187	return res.Err()	1✔
188	}	1✔
189
190	// ProvisionDevice provisions a new device
191	func (db *DataStoreMongo) ProvisionDevice(ctx context.Context, tenantID, deviceID string) error {	4✔
192	coll := db.client.Database(DbName).Collection(DevicesCollectionName)	4✔
193		4✔
194	now := clock.Now().UTC()	4✔
195		4✔
196	updateOpts := &mopts.UpdateOptions{}	4✔
197	updateOpts.SetUpsert(true)	4✔
198	_, err := coll.UpdateOne(ctx,	4✔
199	bson.M{dbFieldID: deviceID, mstore.FieldTenantID: tenantID},	4✔
200	bson.M{	4✔
201	"$setOnInsert": bson.M{	4✔
202	dbFieldStatus: model.DeviceStatusUnknown,	4✔
203	dbFieldCreatedTs: &now,	4✔
204	dbFieldUpdatedTs: &now,	4✔
205	mstore.FieldTenantID: tenantID,	4✔
206	},	4✔
207	},	4✔
208	updateOpts,	4✔
209	)	4✔
210	return err	4✔
211	}	4✔
212
213	// DeleteDevice deletes a device
214	func (db *DataStoreMongo) DeleteDevice(ctx context.Context, tenantID, deviceID string) error {	1✔
215	coll := db.client.Database(DbName).Collection(DevicesCollectionName)	1✔
216		1✔
217	_, err := coll.DeleteOne(ctx, bson.M{dbFieldID: deviceID, mstore.FieldTenantID: tenantID})	1✔
218	return err	1✔
219	}	1✔
220
221	// GetDevice returns a device
222	func (db *DataStoreMongo) GetDevice(
223	ctx context.Context,
224	tenantID string,
225	deviceID string,
226	) (*model.Device, error) {	22✔
227	coll := db.client.Database(DbName).Collection(DevicesCollectionName)	22✔
228		22✔
229	cur := coll.FindOne(ctx, bson.M{dbFieldID: deviceID, mstore.FieldTenantID: tenantID})	22✔
230		22✔
231	device := &model.Device{}	22✔
232	if err := cur.Decode(&device); err != nil {	25✔
233	if err == mongo.ErrNoDocuments {	6✔
234	return nil, nil	3✔
235	}	3✔
236	return nil, err	×
237	}
238
239	return device, nil	19✔
240	}
241
242	// UpsertDeviceStatus upserts the connection status of a device
243	func (db *DataStoreMongo) UpsertDeviceStatus(
244	ctx context.Context,
245	tenantID string,
246	deviceID string,
247	status string,
248	) error {	7✔
249	coll := db.client.Database(DbName).Collection(DevicesCollectionName)	7✔
250		7✔
251	updateOpts := &mopts.UpdateOptions{}	7✔
252	updateOpts.SetUpsert(true)	7✔
253		7✔
254	now := clock.Now().UTC()	7✔
255		7✔
256	_, err := coll.UpdateOne(ctx,	7✔
257	bson.M{dbFieldID: deviceID, mstore.FieldTenantID: tenantID},	7✔
258	bson.M{	7✔
259	"$set": bson.M{	7✔
260	dbFieldStatus: status,	7✔
261	dbFieldUpdatedTs: &now,	7✔
262	},	7✔
263	"$setOnInsert": bson.M{	7✔
264	dbFieldCreatedTs: &now,	7✔
265	mstore.FieldTenantID: tenantID,	7✔
266	},	7✔
267	},	7✔
268	updateOpts,	7✔
269	)	7✔
270		7✔
271	return err	7✔
272	}	7✔
273
274	// AllocateSession allocates a new session.
275	func (db DataStoreMongo) AllocateSession(ctx context.Context, sess model.Session) error {	9✔
276		9✔
277	if err := sess.Validate(); err != nil {	10✔
278	return errors.Wrap(err, "store: cannot allocate invalid Session")	1✔
279	}	1✔
280
281	coll := db.client.Database(DbName).Collection(SessionsCollectionName)	8✔
282	tenantElem := bson.E{Key: mstore.FieldTenantID, Value: sess.TenantID}	8✔
283	_, err := coll.InsertOne(ctx, mdoc.DocumentFromStruct(*sess, tenantElem))	8✔
284	if err != nil {	9✔
285	return errors.Wrap(err, "store: failed to allocate session")	1✔
286	}	1✔
287
288	return nil	7✔
289	}
290
291	// DeleteSession deletes a session
292	func (db *DataStoreMongo) DeleteSession(
293	ctx context.Context, sessionID string,
294	) (*model.Session, error) {	10✔
295	collSess := db.client.Database(DbName).	10✔
296	Collection(SessionsCollectionName)	10✔
297		10✔
298	sess := new(model.Session)	10✔
299	err := collSess.FindOneAndDelete(	10✔
300	ctx, mstore.WithTenantID(ctx, bson.D{{Key: dbFieldID, Value: sessionID}}),	10✔
301	).Decode(sess)	10✔
302	if err != nil {	13✔
303	if err == mongo.ErrNoDocuments {	5✔
304	return nil, store.ErrSessionNotFound	2✔
305	} else {	3✔
306	return nil, err	1✔
307	}	1✔
308	}
309	if idty := identity.FromContext(ctx); idty != nil {	13✔
310	sess.TenantID = idty.Tenant	6✔
311	}	6✔
312	return sess, nil	7✔
313	}
314
315	// GetSession returns a session
316	func (db *DataStoreMongo) GetSession(
317	ctx context.Context,
318	sessionID string,
319	) (*model.Session, error) {	4✔
320	collSess := db.client.	4✔
321	Database(DbName).	4✔
322	Collection(SessionsCollectionName)	4✔
323		4✔
324	session := &model.Session{}	4✔
325	err := collSess.	4✔
326	FindOne(ctx, mstore.WithTenantID(ctx, bson.M{dbFieldID: sessionID})).	4✔
327	Decode(session)	4✔
328	if err != nil {	6✔
329	if err == mongo.ErrNoDocuments {	3✔
330	return nil, store.ErrSessionNotFound	1✔
331	}	1✔
332	return nil, err	1✔
333	}
334	idty := identity.FromContext(ctx)	2✔
335	if idty != nil {	3✔
336	session.TenantID = idty.Tenant	1✔
337	}	1✔
338
339	return session, nil	2✔
340	}
341
342	func sendControlMessage(control app.Control, sessionID string, w io.Writer) (int, error) {	6✔
343	messageType := ""	6✔
344	var data []byte	6✔
345	properties := make(map[string]interface{})	6✔
346		6✔
347	switch control.Type {	6✔
348	case app.DelayMessage:	6✔
349	messageType = model.DelayMessageName	6✔
350	properties[model.DelayMessageValueField] = control.DelayMs	6✔
351	case app.ResizeMessage:	×
352	messageType = shell.MessageTypeResizeShell	×
353	properties[model.ResizeMessageTermHeightField] = control.TerminalHeight	×
354	properties[model.ResizeMessageTermWidthField] = control.TerminalWidth	×
355	default:	×
356	return 0, ErrUnknownControlMessageType	×
357	}
358
359	msg := ws.ProtoMsg{	6✔
360	Header: ws.ProtoHdr{	6✔
361	Proto: ws.ProtoTypeShell,	6✔
362	MsgType: messageType,	6✔
363	SessionID: sessionID,	6✔
364	Properties: properties,	6✔
365	},	6✔
366	Body: data,	6✔
367	}	6✔
368	messagePacked, err := msgpack.Marshal(&msg)	6✔
369	if err != nil {	6✔
370	return 0, err	×
371	} else {	6✔
372	return w.Write(messagePacked)	6✔
373	}	6✔
374	}
375
376	// GetSession writes session recordings to given io.Writer
377	func (db *DataStoreMongo) WriteSessionRecords(ctx context.Context,
378	sessionID string,
379	w io.Writer) error {	2✔
380	l := log.FromContext(ctx)	2✔
381	collRecording := db.client.Database(DbName).	2✔
382	Collection(RecordingsCollectionName)	2✔
383	collControl := db.client.Database(DbName).	2✔
384	Collection(ControlCollectionName)	2✔
385		2✔
386	findOptions := mopts.Find()	2✔
387	sortField := bson.M{	2✔
388	"created_ts": 1,	2✔
389	}	2✔
390	findOptions.SetSort(sortField)	2✔
391	recordingsCursor, err := collRecording.Find(ctx,	2✔
392	mstore.WithTenantID(ctx, bson.M{	2✔
393	dbFieldSessionID: sessionID,	2✔
394	}),	2✔
395	findOptions,	2✔
396	)	2✔
397	if err != nil {	2✔
398	return err	×
399	}	×
400	defer recordingsCursor.Close(ctx)	2✔
401		2✔
402	controlCursor, err := collControl.Find(ctx,	2✔
403	mstore.WithTenantID(ctx, bson.M{	2✔
404	dbFieldSessionID: sessionID,	2✔
405	}),	2✔
406	findOptions,	2✔
407	)	2✔
408	if err != nil {	2✔
409	return err	×
410	}	×
411	defer controlCursor.Close(ctx)	2✔
412		2✔
413	controlReader := NewControlMessageReader(ctx, controlCursor)	2✔
414	recordingReader := NewRecordingReader(ctx, recordingsCursor)	2✔
415		2✔
416	recordingWriter := NewRecordingWriter(sessionID, w)	2✔
417	recordingBuffer := make([]byte, recordingReadBufferSize)	2✔
418	recordingBytesSent := 0	2✔
419	for {	10✔
420	control := controlReader.Pop()	8✔
421	if control == nil {	10✔
422	l.Debug("WriteSessionRecords: no more control " +	2✔
423	"messages, flushing the recording upstream.")	2✔
424	//no more control messages, we send the whole recording	2✔
425	n, err := io.Copy(recordingWriter, recordingReader)	2✔
426	if err != nil && err != io.ErrShortWrite && n < 1 {	2✔
427	l.Errorf("WriteSessionRecords: "+	×
428	"error writing recording data, err: %+v n:%d",	×
429	err, n)	×
430	}	×
431	if n == 0 {	2✔
432	l.Errorf("WriteSessionRecords: "+	×
433	"failed to write any recording data, err: %+v",	×
434	err)	×
435	} else {	2✔
436	recordingBytesSent += int(n)	2✔
437	}	2✔
438	break	2✔
439	} else {	6✔
440	if recordingBytesSent > control.Offset {	6✔
441	//this should never happen, we missed	×
442	//the control message, data inconsistency	×
443	l.Errorf("WriteSessionRecords: recordingBytesSent > control.Offset")	×
444	err = ErrRecordingDataInconsistent	×
445	break	×
446	}
447
448	bytesUntilControlMessage := control.Offset - recordingBytesSent	6✔
449	l.Debugf("(1) WriteSessionRecords: control.Offset:%d"+	6✔
450	" recordingBytesSent:%d "+	6✔
451	" bytesUntilControlMessage:%d (recordingBuffer.len=%d) "+	6✔
452	"reading up to %d bytes of recording and sending.",	6✔
453	control.Offset,	6✔
454	recordingBytesSent,	6✔
455	bytesUntilControlMessage,	6✔
456	len(recordingBuffer),	6✔
457	control.Offset-recordingBytesSent)	6✔
458	//it is possible that the recording is larger than one recordingBuffer,	6✔
459	//we need to send until we have the control.Offset in the buffer	6✔
460	for bytesUntilControlMessage > len(recordingBuffer) {	6✔
461	n, e := recordingReader.Read(recordingBuffer)	×
462	if n > 0 {	×
463	_, err = sendRecordingMessage(recordingBuffer[:n],	×
464	sessionID,	×
465	w)	×
466	if err != nil {	×
467	l.Errorf("error sending recording data: %s",	×
468	err.Error())	×
469	break	×
470	}
471	recordingBytesSent += n	×
472	bytesUntilControlMessage = control.Offset -	×
473	recordingBytesSent	×
474	}
475	if e != nil \|\| n == 0 {	×
476	break	×
477	}
478	}
479	if err != nil {	6✔
480	break	×
481	}
482
483	bytesUntilControlMessage = control.Offset - recordingBytesSent	6✔
484	l.Debugf("(2) WriteSessionRecords: control.Offset:%d"+	6✔
485	" recordingBytesSent:%d "+	6✔
486	" bytesUntilControlMessage:%d (recordingBuffer.len=%d) "+	6✔
487	"reading up to %d bytes of recording and sending.",	6✔
488	control.Offset,	6✔
489	recordingBytesSent,	6✔
490	bytesUntilControlMessage,	6✔
491	len(recordingBuffer),	6✔
492	bytesUntilControlMessage)	6✔
493	//this means that the control offset is in the future	6✔
494	//part of the recording buffer	6✔
495	//we can send up to control.Offset-recordingBytesSent	6✔
496	//bytes and then send the control message	6✔
497	n, e := recordingReader.Read(recordingBuffer[:bytesUntilControlMessage])	6✔
498	l.Debugf("recordingReader.Read(len=%d)=%d,%+v",	6✔
499	bytesUntilControlMessage, n, e)	6✔
500	if n > 0 {	12✔
501	_, err = sendRecordingMessage(recordingBuffer[:n], sessionID, w)	6✔
502	if err != nil {	6✔
503	l.Errorf("error sending recording data: %s",	×
504	err.Error())	×
505	break	×
506	}
507	recordingBytesSent += n	6✔
508	}
509	l.Debugf("WriteSessionRecords: sending %+v.", *control)	6✔
510	_, err = sendControlMessage(*control, sessionID, w)	6✔
511	if err != nil {	6✔
512	l.Errorf("error sending recording data: %s",	×
513	err.Error())	×
514	break	×
515	}
516	}
517	}
518	l.Infof("session playback: WriteSessionRecords: sent %d bytes.", recordingBytesSent)	2✔
519		2✔
520	return err	2✔
521	}
522
523	// SetSession saves a session recording
524	func (db *DataStoreMongo) InsertSessionRecording(ctx context.Context,
525	sessionID string,
526	sessionBytes []byte) error {	4✔
527	coll := db.client.Database(DbName).Collection(RecordingsCollectionName)	4✔
528		4✔
529	now := clock.Now().UTC()	4✔
530	recording := model.Recording{	4✔
531	ID: uuid.New(),	4✔
532	SessionID: sessionID,	4✔
533	Recording: sessionBytes,	4✔
534	CreatedTs: now,	4✔
535	ExpireTs: now.Add(db.recordingExpire),	4✔
536	}	4✔
537	_, err := coll.InsertOne(ctx,	4✔
538	mstore.WithTenantID(ctx, &recording),	4✔
539	)	4✔
540	return err	4✔
541	}	4✔
542
543	// Inserts control data recording
544	func (db *DataStoreMongo) InsertControlRecording(ctx context.Context,
545	sessionID string,
546	sessionBytes []byte) error {	×
547	coll := db.client.Database(DbName).	×
548	Collection(ControlCollectionName)	×
549		×
550	now := clock.Now().UTC()	×
551	recording := model.ControlData{	×
552	ID: uuid.New(),	×
553	SessionID: sessionID,	×
554	Control: sessionBytes,	×
555	CreatedTs: now,	×
556	ExpireTs: now.Add(db.recordingExpire),	×
557	}	×
558	_, err := coll.InsertOne(ctx,	×
559	mstore.WithTenantID(ctx, &recording),	×
560	)	×
561	return err	×
562	}	×
563
564	// Close disconnects the client
565	func (db *DataStoreMongo) Close() error {	×
566	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)	×
567	defer cancel()	×
568	err := db.client.Disconnect(ctx)	×
569	return err	×
570	}	×
571
572	//nolint:unused
573	func (db *DataStoreMongo) DropDatabase() error {	27✔
574	ctx := context.Background()	27✔
575	err := db.client.Database(DbName).Drop(ctx)	27✔
576	return err	27✔
577	}	27✔

mendersoftware / deviceconnect / 938326290

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