826658230

Build Type

gitlab-ci

Committed by Peter Grzybowski

Commit Message

chore: moving to single db

Pull Request Pull Request #638: chore: moving to single db

Run Details

334 of 405 new or added lines in 5 files covered. (82.47%)

38 existing lines in 3 files now uncovered.

4669 of 5588 relevant lines covered (83.55%)

75.19 hits per line

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

/store/mongo/migration_2_0_0.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 mongo

import (
        "context"
        "strings"

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

        "github.com/mendersoftware/go-lib-micro/identity"
        "github.com/mendersoftware/go-lib-micro/mongo/migrate"
        "github.com/mendersoftware/go-lib-micro/mongo/oid"
        mstorev1 "github.com/mendersoftware/go-lib-micro/store"
        mstore "github.com/mendersoftware/go-lib-micro/store/v2"
        "go.mongodb.org/mongo-driver/bson"
)

const (
        findBatchSize = 255
)

type migration_2_0_0 struct {
        ds  *DataStoreMongo
        ctx context.Context
}

var DbDevicesCollectionIndices = []mongo.IndexModel{
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldID, Value: 1},
                },
                Options: mopts.Index().
                        SetName(mstore.FieldTenantID + "_" + dbFieldID).
                        SetUnique(true),
        },
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldStatus, Value: 1},
                },
                Options: mopts.Index().
                        SetName(mstore.FieldTenantID + "_" + dbFieldStatus),
        },
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldIDDataSha, Value: 1},
                },
                //nolint:staticcheck // SA1019
                Options: mopts.Index().
                        SetName(strings.Join([]string{
                                mstore.FieldTenantID,
                                dbFieldIDDataSha,
                        }, "_")).
                        SetUnique(true).
                        SetBackground(false),
        },
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldStatus, Value: 1},
                        {Key: dbFieldID, Value: 1},
                },
                //nolint:staticcheck // SA1019
                Options: mopts.Index().
                        SetName(strings.Join([]string{
                                mstore.FieldTenantID,
                                dbFieldStatus,
                                dbFieldID,
                        }, "_")).
                        SetBackground(false),
        },
}

var DbAuthSetsCollectionIndices = []mongo.IndexModel{
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldID, Value: 1},
                },
                Options: mopts.Index().
                        SetName(mstore.FieldTenantID + "_" + dbFieldID).
                        SetUnique(true),
        },
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldDeviceID, Value: 1},
                        {Key: dbFieldIDDataSha, Value: 1},
                        {Key: dbFieldPubKey, Value: 1},
                },
                Options: mopts.Index().
                        SetName(strings.Join([]string{
                                mstore.FieldTenantID,
                                dbFieldDeviceID,
                                dbFieldIDDataSha,
                                dbFieldPubKey,
                        }, "_")).
                        SetUnique(true),
        },
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldIDDataSha, Value: 1},
                        {Key: dbFieldPubKey, Value: 1},
                },
                Options: mopts.Index().
                        SetName(strings.Join([]string{
                                mstore.FieldTenantID,
                                dbFieldIDDataSha,
                                dbFieldPubKey,
                        }, "_")).
                        SetUnique(true),
        },
}

var DbLimitsCollectionIndices = []mongo.IndexModel{
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldID, Value: 1},
                },
                Options: mopts.Index().
                        SetName(mstore.FieldTenantID + "_" + dbFieldID),
        },
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldName, Value: 1},
                },
                Options: mopts.Index().
                        SetName(mstore.FieldTenantID + "_" + dbFieldName).
                        SetUnique(true),
        },
}

var DbTokensCollectionIndices = []mongo.IndexModel{
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldID, Value: 1},
                },
                Options: mopts.Index().
                        SetName(mstore.FieldTenantID + "_" + dbFieldID),
        },
        {
                Keys: bson.D{
                        {Key: mstore.FieldTenantID, Value: 1},
                        {Key: dbFieldExpTime, Value: 1},
                },
                Options: mopts.Index().
                        SetName(mstore.FieldTenantID + "_" + dbFieldExpTime),
        },
}

// Up creates an index on status and id in the devices collection
func (m *migration_2_0_0) Up(from migrate.Version) error {
        ctx := context.Background()
        client := m.ds.client

        collections := map[string]struct {
                Indexes []mongo.IndexModel
        }{
                DbAuthSetColl: {
                        Indexes: DbAuthSetsCollectionIndices,
                },
                DbDevicesColl: {
                        Indexes: DbDevicesCollectionIndices,
                },
                DbLimitsColl: {
                        Indexes: DbLimitsCollectionIndices,
                },
                DbTokensColl: {
                        Indexes: DbTokensCollectionIndices,
                },
        }

        databaseName := mstorev1.DbFromContext(m.ctx, DbName)
        tenantID := mstorev1.TenantFromDbName(databaseName, DbName)
        ctx = identity.WithContext(ctx, &identity.Identity{
                Tenant: tenantID,
        })
        writes := make([]mongo.WriteModel, 0, findBatchSize)

        for collection, idxes := range collections {
                writes = writes[:0]
                findOptions := mopts.Find().
                        SetBatchSize(findBatchSize).
                        SetSort(bson.D{{Key: dbFieldID, Value: 1}})
                collOut := client.Database(DbName).Collection(collection)
                if databaseName == DbName {
                        indices := collOut.Indexes()
                        _, _ = indices.DropAll(ctx)

                        if len(idxes.Indexes) > 0 {
                                _, err := collOut.Indexes().CreateMany(ctx, collections[collection].Indexes)
                                if err != nil {
                                        return err
                                }
                        }
                        _, err := collOut.UpdateMany(ctx, bson.D{
                                {Key: mstore.FieldTenantID, Value: bson.D{
                                        {Key: "$exists", Value: false},
                                }},
                        }, bson.D{{Key: "$set", Value: bson.D{
                                {Key: mstore.FieldTenantID, Value: ""},
                        }}},
                        )
                        if err != nil {
                                return err
                        }
                        continue
                }

                coll := client.Database(databaseName).Collection(collection)
                // get all the documents in the collection
                cur, err := coll.Find(ctx, bson.D{}, findOptions)
                if err != nil {
                        return err
                }
                defer cur.Close(ctx)

                // migrate the documents
                if collection == DbLimitsColl {
                        for cur.Next(ctx) {
                                id := cur.Current.Lookup(dbFieldID)
                                var currentId string
                                err = id.Unmarshal(&currentId)
                                if err != nil {
                                        return errors.Wrap(err, "the id found is un-parsable")
                                }
                                var item bson.D
                                if err = cur.Decode(&item); err != nil {
                                        return err
                                }
                                item = append(item, bson.E{
                                        Key:   dbFieldName,
                                        Value: currentId,
                                })
                                item = item[1:]
                                item = findAndReplace(item, dbFieldID, oid.NewUUIDv4().String())

                                writes = append(writes, mongo.
                                        NewReplaceOneModel().
                                        SetFilter(bson.D{{Key: dbFieldName, Value: id}}).
                                        SetUpsert(true).
                                        SetReplacement(mstore.WithTenantID(ctx, item)))
                                if len(writes) == findBatchSize {
                                        _, err = collOut.BulkWrite(ctx, writes)
                                        if err != nil {
                                                return err
                                        }
                                        writes = writes[:0]
                                }
                        }
                } else {
                        for cur.Next(ctx) {
                                id := cur.Current.Lookup(dbFieldID)
                                var item bson.D
                                if err = cur.Decode(&item); err != nil {
                                        return err
                                }
                                writes = append(writes, mongo.
                                        NewReplaceOneModel().
                                        SetFilter(bson.D{{Key: dbFieldID, Value: id}}).
                                        SetUpsert(true).
                                        SetReplacement(mstore.WithTenantID(ctx, item)))
                                if len(writes) == findBatchSize {
                                        _, err = collOut.BulkWrite(ctx, writes)
                                        if err != nil {
                                                return err
                                        }
                                        writes = writes[:0]
                                }
                        }
                }
                if len(writes) > 0 {
                        _, err := collOut.BulkWrite(ctx, writes)
                        if err != nil {
                                return err
                        }
                }
        }

        return nil
}

func findAndReplace(item bson.D, key string, value string) bson.D {
        for i, d := range item {
                if d.Key == key {
                        item[i].Value = value
                }
        }
        return item
}

func (m *migration_2_0_0) Version() migrate.Version {
        return migrate.MakeVersion(2, 0, 0)
}

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	package mongo
15
16	import (
17	"context"
18	"strings"
19
20	"github.com/pkg/errors"
21	"go.mongodb.org/mongo-driver/mongo"
22	mopts "go.mongodb.org/mongo-driver/mongo/options"
23
24	"github.com/mendersoftware/go-lib-micro/identity"
25	"github.com/mendersoftware/go-lib-micro/mongo/migrate"
26	"github.com/mendersoftware/go-lib-micro/mongo/oid"
27	mstorev1 "github.com/mendersoftware/go-lib-micro/store"
28	mstore "github.com/mendersoftware/go-lib-micro/store/v2"
29	"go.mongodb.org/mongo-driver/bson"
30	)
31
32	const (
33	findBatchSize = 255
34	)
35
36	type migration_2_0_0 struct {
37	ds *DataStoreMongo
38	ctx context.Context
39	}
40
41	var DbDevicesCollectionIndices = []mongo.IndexModel{
42	{
43	Keys: bson.D{
44	{Key: mstore.FieldTenantID, Value: 1},
45	{Key: dbFieldID, Value: 1},
46	},
47	Options: mopts.Index().
48	SetName(mstore.FieldTenantID + "_" + dbFieldID).
49	SetUnique(true),
50	},
51	{
52	Keys: bson.D{
53	{Key: mstore.FieldTenantID, Value: 1},
54	{Key: dbFieldStatus, Value: 1},
55	},
56	Options: mopts.Index().
57	SetName(mstore.FieldTenantID + "_" + dbFieldStatus),
58	},
59	{
60	Keys: bson.D{
61	{Key: mstore.FieldTenantID, Value: 1},
62	{Key: dbFieldIDDataSha, Value: 1},
63	},
64	//nolint:staticcheck // SA1019
65	Options: mopts.Index().
66	SetName(strings.Join([]string{
67	mstore.FieldTenantID,
68	dbFieldIDDataSha,
69	}, "_")).
70	SetUnique(true).
71	SetBackground(false),
72	},
73	{
74	Keys: bson.D{
75	{Key: mstore.FieldTenantID, Value: 1},
76	{Key: dbFieldStatus, Value: 1},
77	{Key: dbFieldID, Value: 1},
78	},
79	//nolint:staticcheck // SA1019
80	Options: mopts.Index().
81	SetName(strings.Join([]string{
82	mstore.FieldTenantID,
83	dbFieldStatus,
84	dbFieldID,
85	}, "_")).
86	SetBackground(false),
87	},
88	}
89
90	var DbAuthSetsCollectionIndices = []mongo.IndexModel{
91	{
92	Keys: bson.D{
93	{Key: mstore.FieldTenantID, Value: 1},
94	{Key: dbFieldID, Value: 1},
95	},
96	Options: mopts.Index().
97	SetName(mstore.FieldTenantID + "_" + dbFieldID).
98	SetUnique(true),
99	},
100	{
101	Keys: bson.D{
102	{Key: mstore.FieldTenantID, Value: 1},
103	{Key: dbFieldDeviceID, Value: 1},
104	{Key: dbFieldIDDataSha, Value: 1},
105	{Key: dbFieldPubKey, Value: 1},
106	},
107	Options: mopts.Index().
108	SetName(strings.Join([]string{
109	mstore.FieldTenantID,
110	dbFieldDeviceID,
111	dbFieldIDDataSha,
112	dbFieldPubKey,
113	}, "_")).
114	SetUnique(true),
115	},
116	{
117	Keys: bson.D{
118	{Key: mstore.FieldTenantID, Value: 1},
119	{Key: dbFieldIDDataSha, Value: 1},
120	{Key: dbFieldPubKey, Value: 1},
121	},
122	Options: mopts.Index().
123	SetName(strings.Join([]string{
124	mstore.FieldTenantID,
125	dbFieldIDDataSha,
126	dbFieldPubKey,
127	}, "_")).
128	SetUnique(true),
129	},
130	}
131
132	var DbLimitsCollectionIndices = []mongo.IndexModel{
133	{
134	Keys: bson.D{
135	{Key: mstore.FieldTenantID, Value: 1},
136	{Key: dbFieldID, Value: 1},
137	},
138	Options: mopts.Index().
139	SetName(mstore.FieldTenantID + "_" + dbFieldID),
140	},
141	{
142	Keys: bson.D{
143	{Key: mstore.FieldTenantID, Value: 1},
144	{Key: dbFieldName, Value: 1},
145	},
146	Options: mopts.Index().
147	SetName(mstore.FieldTenantID + "_" + dbFieldName).
148	SetUnique(true),
149	},
150	}
151
152	var DbTokensCollectionIndices = []mongo.IndexModel{
153	{
154	Keys: bson.D{
155	{Key: mstore.FieldTenantID, Value: 1},
156	{Key: dbFieldID, Value: 1},
157	},
158	Options: mopts.Index().
159	SetName(mstore.FieldTenantID + "_" + dbFieldID),
160	},
161	{
162	Keys: bson.D{
163	{Key: mstore.FieldTenantID, Value: 1},
164	{Key: dbFieldExpTime, Value: 1},
165	},
166	Options: mopts.Index().
167	SetName(mstore.FieldTenantID + "_" + dbFieldExpTime),
168	},
169	}
170
171	// Up creates an index on status and id in the devices collection
172	func (m *migration_2_0_0) Up(from migrate.Version) error {	205✔
173	ctx := context.Background()	205✔
174	client := m.ds.client	205✔
175		205✔
176	collections := map[string]struct {	205✔
177	Indexes []mongo.IndexModel	205✔
178	}{	205✔
179	DbAuthSetColl: {	205✔
180	Indexes: DbAuthSetsCollectionIndices,	205✔
181	},	205✔
182	DbDevicesColl: {	205✔
183	Indexes: DbDevicesCollectionIndices,	205✔
184	},	205✔
185	DbLimitsColl: {	205✔
186	Indexes: DbLimitsCollectionIndices,	205✔
187	},	205✔
188	DbTokensColl: {	205✔
189	Indexes: DbTokensCollectionIndices,	205✔
190	},	205✔
191	}	205✔
192		205✔
193	databaseName := mstorev1.DbFromContext(m.ctx, DbName)	205✔
194	tenantID := mstorev1.TenantFromDbName(databaseName, DbName)	205✔
195	ctx = identity.WithContext(ctx, &identity.Identity{	205✔
196	Tenant: tenantID,	205✔
197	})	205✔
198	writes := make([]mongo.WriteModel, 0, findBatchSize)	205✔
199		205✔
200	for collection, idxes := range collections {	1,022✔
201	writes = writes[:0]	817✔
202	findOptions := mopts.Find().	817✔
203	SetBatchSize(findBatchSize).	817✔
204	SetSort(bson.D{{Key: dbFieldID, Value: 1}})	817✔
205	collOut := client.Database(DbName).Collection(collection)	817✔
206	if databaseName == DbName {	1,346✔
207	indices := collOut.Indexes()	529✔
208	_, _ = indices.DropAll(ctx)	529✔
209		529✔
210	if len(idxes.Indexes) > 0 {	1,058✔
211	_, err := collOut.Indexes().CreateMany(ctx, collections[collection].Indexes)	529✔
212	if err != nil {	529✔
NEW 213	return err	×
NEW 214	}	×
215	}
216	_, err := collOut.UpdateMany(ctx, bson.D{	529✔
217	{Key: mstore.FieldTenantID, Value: bson.D{	529✔
218	{Key: "$exists", Value: false},	529✔
219	}},	529✔
220	}, bson.D{{Key: "$set", Value: bson.D{	529✔
221	{Key: mstore.FieldTenantID, Value: ""},	529✔
222	}}},	529✔
223	)	529✔
224	if err != nil {	529✔
NEW 225	return err	×
NEW 226	}	×
227	continue	529✔
228	}
229
230	coll := client.Database(databaseName).Collection(collection)	289✔
231	// get all the documents in the collection	289✔
232	cur, err := coll.Find(ctx, bson.D{}, findOptions)	289✔
233	if err != nil {	289✔
NEW 234	return err	×
NEW 235	}	×
236	defer cur.Close(ctx)	289✔
237		289✔
238	// migrate the documents	289✔
239	if collection == DbLimitsColl {	362✔
240	for cur.Next(ctx) {	73✔
NEW 241	id := cur.Current.Lookup(dbFieldID)	×
NEW 242	var currentId string	×
NEW 243	err = id.Unmarshal(&currentId)	×
NEW 244	if err != nil {	×
NEW 245	return errors.Wrap(err, "the id found is un-parsable")	×
NEW 246	}	×
NEW 247	var item bson.D	×
NEW 248	if err = cur.Decode(&item); err != nil {	×
NEW 249	return err	×
NEW 250	}	×
NEW 251	item = append(item, bson.E{	×
NEW 252	Key: dbFieldName,	×
NEW 253	Value: currentId,	×
NEW 254	})	×
NEW 255	item = item[1:]	×
NEW 256	item = findAndReplace(item, dbFieldID, oid.NewUUIDv4().String())	×
NEW 257		×
NEW 258	writes = append(writes, mongo.	×
NEW 259	NewReplaceOneModel().	×
NEW 260	SetFilter(bson.D{{Key: dbFieldName, Value: id}}).	×
NEW 261	SetUpsert(true).	×
NEW 262	SetReplacement(mstore.WithTenantID(ctx, item)))	×
NEW 263	if len(writes) == findBatchSize {	×
NEW 264	_, err = collOut.BulkWrite(ctx, writes)	×
NEW 265	if err != nil {	×
NEW 266	return err	×
NEW 267	}	×
NEW 268	writes = writes[:0]	×
269	}
270	}
271	} else {	217✔
272	for cur.Next(ctx) {	10,757✔
273	id := cur.Current.Lookup(dbFieldID)	10,540✔
274	var item bson.D	10,540✔
275	if err = cur.Decode(&item); err != nil {	10,540✔
NEW 276	return err	×
NEW 277	}	×
278	writes = append(writes, mongo.	10,540✔
279	NewReplaceOneModel().	10,540✔
280	SetFilter(bson.D{{Key: dbFieldID, Value: id}}).	10,540✔
281	SetUpsert(true).	10,540✔
282	SetReplacement(mstore.WithTenantID(ctx, item)))	10,540✔
283	if len(writes) == findBatchSize {	10,540✔
NEW 284	_, err = collOut.BulkWrite(ctx, writes)	×
NEW 285	if err != nil {	×
NEW 286	return err	×
NEW 287	}	×
NEW 288	writes = writes[:0]	×
289	}
290	}
291	}
292	if len(writes) > 0 {	357✔
293	_, err := collOut.BulkWrite(ctx, writes)	68✔
294	if err != nil {	68✔
NEW 295	return err	×
NEW 296	}	×
297	}
298	}
299
300	return nil	205✔
301	}
302
NEW 303	func findAndReplace(item bson.D, key string, value string) bson.D {	×
NEW 304	for i, d := range item {	×
NEW 305	if d.Key == key {	×
NEW 306	item[i].Value = value	×
NEW 307	}	×
308	}
NEW 309	return item	×
310	}
311
312	func (m *migration_2_0_0) Version() migrate.Version {	281✔
313	return migrate.MakeVersion(2, 0, 0)	281✔
314	}	281✔

mendersoftware / deviceauth / 826658230

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