734966968

Build Type

gitlab-ci

Committed by Fabio Tranchitella

Commit Message

feat: add support to index the device deployment data in the indexer

Pull Request Pull Request #87: MEN-5930: index device deployment objects

Run Details

734 of 987 new or added lines in 17 files covered. (74.37%)

4 existing lines in 2 files now uncovered.

2556 of 3139 relevant lines covered (81.43%)

16.17 hits per line

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

71.17

/app/reporting/reporting_deployments.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 reporting

import (
        "context"
        "encoding/json"

        "github.com/pkg/errors"

        "github.com/mendersoftware/reporting/model"
)

// AggregateDeployments aggregates deployments data
func (app *app) AggregateDeployments(
        ctx context.Context,
        aggregateParams *model.AggregateDeploymentsParams,
) ([]model.DeviceAggregation, error) {
        searchParams := &model.DeploymentsSearchParams{
                Filters:  aggregateParams.Filters,
                TenantID: aggregateParams.TenantID,
        }
        query, err := model.BuildDeploymentsQuery(*searchParams)
        if err != nil {
                return nil, err
        }
        if searchParams.TenantID != "" {
                query = query.Must(model.M{
                        "term": model.M{
                                model.FieldNameTenantID: searchParams.TenantID,
                        },
                })
        }

        aggregations, err := model.BuildDeploymentsAggregations(aggregateParams.Aggregations)
        if err != nil {
                return nil, err
        }

        query = query.WithSize(0).With(map[string]interface{}{
                "aggs": aggregations,
        })
        esRes, err := app.store.AggregateDeployments(ctx, query)
        if err != nil {
                return nil, err
        }

        aggregationsS, ok := esRes["aggregations"].(map[string]interface{})
        if !ok {
                return nil, errors.New("can't process store aggregations slice")
        }
        res, err := app.storeToDeviceAggregations(ctx, searchParams.TenantID, aggregationsS)
        if err != nil {
                return nil, err
        }

        return res, nil
}

// SearchDevices searches device data
func (app *app) SearchDeployments(
        ctx context.Context,
        searchParams *model.DeploymentsSearchParams,
) ([]model.Deployment, int, error) {
        query, err := model.BuildDeploymentsQuery(*searchParams)
        if err != nil {
                return nil, 0, err
        }

        if searchParams.TenantID != "" {
                query = query.Must(model.M{
                        "term": model.M{
                                model.FieldNameTenantID: searchParams.TenantID,
                        },
                })
        }

        if len(searchParams.DeviceIDs) > 0 {
                query = query.Must(model.M{
                        "terms": model.M{
                                model.FieldNameID: searchParams.DeviceIDs,
                        },
                })
        }

        esRes, err := app.store.SearchDeployments(ctx, query)
        if err != nil {
                return nil, 0, err
        }

        res, total, err := app.storeToDeployments(ctx, searchParams.TenantID, esRes)
        if err != nil {
                return nil, 0, err
        }

        return res, total, err
}

// storeToInventoryDevs translates ES results directly to inventory devices
func (a *app) storeToDeployments(
        ctx context.Context, tenantID string, storeRes map[string]interface{},
) ([]model.Deployment, int, error) {
        depls := []model.Deployment{}

        hitsM, ok := storeRes["hits"].(map[string]interface{})
        if !ok {
                return nil, 0, errors.New("can't process store hits map")
        }

        hitsTotalM, ok := hitsM["total"].(map[string]interface{})
        if !ok {
                return nil, 0, errors.New("can't process total hits struct")
        }

        total, ok := hitsTotalM["value"].(float64)
        if !ok {
                return nil, 0, errors.New("can't process total hits value")
        }

        hitsS, ok := hitsM["hits"].([]interface{})
        if !ok {
                return nil, 0, errors.New("can't process store hits slice")
        }

        for _, v := range hitsS {
                res, err := a.storeToDeployment(ctx, tenantID, v)
                if err != nil {
                        return nil, 0, err
                }

                depls = append(depls, *res)
        }

        return depls, int(total), nil
}

func (a *app) storeToDeployment(ctx context.Context, tenantID string,
        storeRes interface{}) (*model.Deployment, error) {
        resM, ok := storeRes.(map[string]interface{})
        if !ok {
                return nil, errors.New("can't process individual hit")
        }

        // if query has a 'fields' clause, use 'fields' instead of '_source'
        sourceM, ok := resM["_source"].(map[string]interface{})
        if !ok {
                sourceM, ok = resM["fields"].(map[string]interface{})
                if !ok {
                        return nil, errors.New("can't process hit's '_source' nor 'fields'")
                }
        }

        source, err := json.Marshal(sourceM)
        if err != nil {
                return nil, errors.Wrap(err, "unable to marshal result into JSON")
        }

        ret := &model.Deployment{}
        err = json.Unmarshal(source, ret)
        if err != nil {
                return nil, errors.Wrap(err, "unable to unmarshal result from JSON")
        }

        return ret, nil
}

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 reporting
16
17	import (
18	"context"
19	"encoding/json"
20
21	"github.com/pkg/errors"
22
23	"github.com/mendersoftware/reporting/model"
24	)
25
26	// AggregateDeployments aggregates deployments data
27	func (app *app) AggregateDeployments(
28	ctx context.Context,
29	aggregateParams *model.AggregateDeploymentsParams,
30	) ([]model.DeviceAggregation, error) {	4✔
31	searchParams := &model.DeploymentsSearchParams{	4✔
32	Filters: aggregateParams.Filters,	4✔
33	TenantID: aggregateParams.TenantID,	4✔
34	}	4✔
35	query, err := model.BuildDeploymentsQuery(*searchParams)	4✔
36	if err != nil {	4✔
NEW 37	return nil, err	×
NEW 38	}	×
39	if searchParams.TenantID != "" {	8✔
40	query = query.Must(model.M{	4✔
41	"term": model.M{	4✔
42	model.FieldNameTenantID: searchParams.TenantID,	4✔
43	},	4✔
44	})	4✔
45	}	4✔
46
47	aggregations, err := model.BuildDeploymentsAggregations(aggregateParams.Aggregations)	4✔
48	if err != nil {	4✔
NEW 49	return nil, err	×
NEW 50	}	×
51
52	query = query.WithSize(0).With(map[string]interface{}{	4✔
53	"aggs": aggregations,	4✔
54	})	4✔
55	esRes, err := app.store.AggregateDeployments(ctx, query)	4✔
56	if err != nil {	4✔
NEW 57	return nil, err	×
NEW 58	}	×
59
60	aggregationsS, ok := esRes["aggregations"].(map[string]interface{})	4✔
61	if !ok {	4✔
NEW 62	return nil, errors.New("can't process store aggregations slice")	×
NEW 63	}	×
64	res, err := app.storeToDeviceAggregations(ctx, searchParams.TenantID, aggregationsS)	4✔
65	if err != nil {	4✔
NEW 66	return nil, err	×
NEW 67	}	×
68
69	return res, nil	4✔
70	}
71
72	// SearchDevices searches device data
73	func (app *app) SearchDeployments(
74	ctx context.Context,
75	searchParams *model.DeploymentsSearchParams,
76	) ([]model.Deployment, int, error) {	12✔
77	query, err := model.BuildDeploymentsQuery(*searchParams)	12✔
78	if err != nil {	14✔
79	return nil, 0, err	2✔
80	}	2✔
81
82	if searchParams.TenantID != "" {	10✔
NEW 83	query = query.Must(model.M{	×
NEW 84	"term": model.M{	×
NEW 85	model.FieldNameTenantID: searchParams.TenantID,	×
NEW 86	},	×
NEW 87	})	×
NEW 88	}	×
89
90	if len(searchParams.DeviceIDs) > 0 {	14✔
91	query = query.Must(model.M{	4✔
92	"terms": model.M{	4✔
93	model.FieldNameID: searchParams.DeviceIDs,	4✔
94	},	4✔
95	})	4✔
96	}	4✔
97
98	esRes, err := app.store.SearchDeployments(ctx, query)	10✔
99	if err != nil {	12✔
100	return nil, 0, err	2✔
101	}	2✔
102
103	res, total, err := app.storeToDeployments(ctx, searchParams.TenantID, esRes)	8✔
104	if err != nil {	10✔
105	return nil, 0, err	2✔
106	}	2✔
107
108	return res, total, err	6✔
109	}
110
111	// storeToInventoryDevs translates ES results directly to inventory devices
112	func (a *app) storeToDeployments(
113	ctx context.Context, tenantID string, storeRes map[string]interface{},
114	) ([]model.Deployment, int, error) {	8✔
115	depls := []model.Deployment{}	8✔
116		8✔
117	hitsM, ok := storeRes["hits"].(map[string]interface{})	8✔
118	if !ok {	8✔
NEW 119	return nil, 0, errors.New("can't process store hits map")	×
NEW 120	}	×
121
122	hitsTotalM, ok := hitsM["total"].(map[string]interface{})	8✔
123	if !ok {	8✔
NEW 124	return nil, 0, errors.New("can't process total hits struct")	×
NEW 125	}	×
126
127	total, ok := hitsTotalM["value"].(float64)	8✔
128	if !ok {	10✔
129	return nil, 0, errors.New("can't process total hits value")	2✔
130	}	2✔
131
132	hitsS, ok := hitsM["hits"].([]interface{})	6✔
133	if !ok {	6✔
NEW 134	return nil, 0, errors.New("can't process store hits slice")	×
NEW 135	}	×
136
137	for _, v := range hitsS {	10✔
138	res, err := a.storeToDeployment(ctx, tenantID, v)	4✔
139	if err != nil {	4✔
NEW 140	return nil, 0, err	×
NEW 141	}	×
142
143	depls = append(depls, *res)	4✔
144	}
145
146	return depls, int(total), nil	6✔
147	}
148
149	func (a *app) storeToDeployment(ctx context.Context, tenantID string,
150	storeRes interface{}) (*model.Deployment, error) {	4✔
151	resM, ok := storeRes.(map[string]interface{})	4✔
152	if !ok {	4✔
NEW 153	return nil, errors.New("can't process individual hit")	×
NEW 154	}	×
155
156	// if query has a 'fields' clause, use 'fields' instead of '_source'
157	sourceM, ok := resM["_source"].(map[string]interface{})	4✔
158	if !ok {	6✔
159	sourceM, ok = resM["fields"].(map[string]interface{})	2✔
160	if !ok {	2✔
NEW 161	return nil, errors.New("can't process hit's '_source' nor 'fields'")	×
NEW 162	}	×
163	}
164
165	source, err := json.Marshal(sourceM)	4✔
166	if err != nil {	4✔
NEW 167	return nil, errors.Wrap(err, "unable to marshal result into JSON")	×
NEW 168	}	×
169
170	ret := &model.Deployment{}	4✔
171	err = json.Unmarshal(source, ret)	4✔
172	if err != nil {	4✔
NEW 173	return nil, errors.Wrap(err, "unable to unmarshal result from JSON")	×
NEW 174	}	×
175
176	return ret, nil	4✔
177	}

mendersoftware / reporting / 734966968

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