742310381

Build Type

gitlab-ci

Committed by Fabio Tranchitella

Commit Message

fix: reindex device and deployment for the `already-installed` status

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

Run Details

86 of 119 new or added lines in 2 files covered. (72.27%)

1 existing line in 1 file now uncovered.

6227 of 7964 relevant lines covered (78.19%)

76.42 hits per line

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

91.08

/client/workflows/client.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 workflows

import (
        "bytes"
        "context"
        "encoding/json"
        "io"
        "net/http"
        "strings"
        "time"

        "github.com/mendersoftware/go-lib-micro/config"
        "github.com/mendersoftware/go-lib-micro/identity"
        "github.com/mendersoftware/go-lib-micro/log"
        "github.com/mendersoftware/go-lib-micro/requestid"
        "github.com/mendersoftware/go-lib-micro/rest_utils"
        "github.com/pkg/errors"

        dconfig "github.com/mendersoftware/deployments/config"
        "github.com/mendersoftware/deployments/model"
)

const (
        healthURL                     = "/api/v1/health"
        generateArtifactURL           = "/api/v1/workflow/generate_artifact"
        reindexReportingURL           = "/api/v1/workflow/reindex_reporting"
        reindexReportingDeploymentURL = "/api/v1/workflow/reindex_reporting_deployment"
        defaultTimeout                = 5 * time.Second
)

// Client is the workflows client
//
//go:generate ../../utils/mockgen.sh
type Client interface {
        CheckHealth(ctx context.Context) error
        StartGenerateArtifact(
                ctx context.Context,
                multipartGenerateImageMsg *model.MultipartGenerateImageMsg,
        ) error
        StartReindexReporting(c context.Context, device string) error
        StartReindexReportingDeployment(c context.Context, device, deployment, id string) error
}

// NewClient returns a new workflows client
func NewClient() Client {
        workflowsBaseURL := config.Config.GetString(dconfig.SettingWorkflows)
        return &client{
                baseURL:    workflowsBaseURL,
                httpClient: &http.Client{Timeout: defaultTimeout},
        }
}

type client struct {
        baseURL    string
        httpClient *http.Client
}

func (c *client) CheckHealth(ctx context.Context) error {
        var (
                apiErr rest_utils.ApiError
                client http.Client
        )

        if ctx == nil {
                ctx = context.Background()
        }
        if _, ok := ctx.Deadline(); !ok {
                var cancel context.CancelFunc
                ctx, cancel = context.WithTimeout(ctx, defaultTimeout)
                defer cancel()
        }
        req, _ := http.NewRequestWithContext(
                ctx, "GET", c.baseURL+healthURL, nil,
        )

        rsp, err := client.Do(req)
        if err != nil {
                return err
        }
        defer rsp.Body.Close()
        if rsp.StatusCode >= http.StatusOK && rsp.StatusCode < 300 {
                return nil
        }
        decoder := json.NewDecoder(rsp.Body)
        err = decoder.Decode(&apiErr)
        if err != nil {
                return errors.Errorf("health check HTTP error: %s", rsp.Status)
        }
        return &apiErr
}

func (c *client) StartGenerateArtifact(
        ctx context.Context,
        multipartGenerateImageMsg *model.MultipartGenerateImageMsg,
) error {
        l := log.FromContext(ctx)
        l.Debugf("Submit generate artifact: tenantID=%s, artifactID=%s",
                multipartGenerateImageMsg.TenantID, multipartGenerateImageMsg.ArtifactID)

        workflowsURL := c.baseURL + generateArtifactURL

        payload, _ := json.Marshal(multipartGenerateImageMsg)
        req, err := http.NewRequest("POST", workflowsURL, strings.NewReader(string(payload)))
        if err != nil {
                return err
        }

        res, err := c.httpClient.Do(req)
        if err != nil {
                return errors.Wrapf(err, "failed to start workflow: generate_artifact")
        }
        defer res.Body.Close()
        if res.StatusCode != http.StatusCreated {
                body, err := io.ReadAll(res.Body)
                if err != nil {
                        body = []byte("<failed to read>")
                }
                l.Errorf("generate artifact failed with status %v, response text: %s",
                        res.StatusCode, body)
                return errors.New("failed to start workflow: generate_artifact")
        }
        return nil
}

func (c *client) StartReindexReporting(ctx context.Context, device string) error {
        if _, ok := ctx.Deadline(); !ok {
                var cancel context.CancelFunc
                ctx, cancel = context.WithTimeout(ctx, defaultTimeout)
                defer cancel()
        }
        id := identity.FromContext(ctx)
        if id == nil {
                return errors.New("workflows: context lacking tenant identity")
        }
        wflow := ReindexWorkflow{
                RequestID: requestid.FromContext(ctx),
                TenantID:  id.Tenant,
                DeviceID:  device,
                Service:   ServiceDeployments,
        }
        payload, _ := json.Marshal(wflow)
        req, err := http.NewRequestWithContext(ctx,
                "POST",
                c.baseURL+reindexReportingURL,
                bytes.NewReader(payload),
        )
        if err != nil {
                return errors.Wrap(err, "workflows: error preparing HTTP request")
        }

        req.Header.Set("Content-Type", "application/json")

        rsp, err := c.httpClient.Do(req)
        if err != nil {
                return errors.Wrap(err, "workflows: failed to trigger reporting reindex")
        }
        defer rsp.Body.Close()

        if rsp.StatusCode < 300 {
                return nil
        }

        if rsp.StatusCode == http.StatusNotFound {
                workflowURIparts := strings.Split(reindexReportingURL, "/")
                workflowName := workflowURIparts[len(workflowURIparts)-1]
                return errors.New(`workflows: workflow "` + workflowName + `" not defined`)
        }

        return errors.Errorf(
                "workflows: unexpected HTTP status from workflows service: %s",
                rsp.Status,
        )
}

func (c *client) StartReindexReportingDeployment(ctx context.Context,
        device, deployment, id string) error {
        if _, ok := ctx.Deadline(); !ok {
                var cancel context.CancelFunc
                ctx, cancel = context.WithTimeout(ctx, defaultTimeout)
                defer cancel()
        }
        ident := identity.FromContext(ctx)
        if ident == nil {
                return errors.New("workflows: context lacking tenant identity")
        }
        wflow := ReindexDeploymentWorkflow{
                RequestID:    requestid.FromContext(ctx),
                TenantID:     ident.Tenant,
                DeviceID:     device,
                DeploymentID: deployment,
                ID:           id,
                Service:      ServiceDeployments,
        }
        payload, _ := json.Marshal(wflow)
        req, err := http.NewRequestWithContext(ctx,
                "POST",
                c.baseURL+reindexReportingDeploymentURL,
                bytes.NewReader(payload),
        )
        if err != nil {
                return errors.Wrap(err, "workflows: error preparing HTTP request")
        }

        req.Header.Set("Content-Type", "application/json")

        rsp, err := c.httpClient.Do(req)
        if err != nil {
                return errors.Wrap(err, "workflows: failed to trigger reporting reindex deployment")
        }
        defer rsp.Body.Close()

        if rsp.StatusCode < 300 {
                return nil
        }

        if rsp.StatusCode == http.StatusNotFound {
                workflowURIparts := strings.Split(reindexReportingDeploymentURL, "/")
                workflowName := workflowURIparts[len(workflowURIparts)-1]
                return errors.New(`workflows: workflow "` + workflowName + `" not defined`)
        }

        return errors.Errorf(
                "workflows: unexpected HTTP status from workflows service: %s",
                rsp.Status,
        )
}

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 workflows
16
17	import (
18	"bytes"
19	"context"
20	"encoding/json"
21	"io"
22	"net/http"
23	"strings"
24	"time"
25
26	"github.com/mendersoftware/go-lib-micro/config"
27	"github.com/mendersoftware/go-lib-micro/identity"
28	"github.com/mendersoftware/go-lib-micro/log"
29	"github.com/mendersoftware/go-lib-micro/requestid"
30	"github.com/mendersoftware/go-lib-micro/rest_utils"
31	"github.com/pkg/errors"
32
33	dconfig "github.com/mendersoftware/deployments/config"
34	"github.com/mendersoftware/deployments/model"
35	)
36
37	const (
38	healthURL = "/api/v1/health"
39	generateArtifactURL = "/api/v1/workflow/generate_artifact"
40	reindexReportingURL = "/api/v1/workflow/reindex_reporting"
41	reindexReportingDeploymentURL = "/api/v1/workflow/reindex_reporting_deployment"
42	defaultTimeout = 5 * time.Second
43	)
44
45	// Client is the workflows client
46	//
47	//go:generate ../../utils/mockgen.sh
48	type Client interface {
49	CheckHealth(ctx context.Context) error
50	StartGenerateArtifact(
51	ctx context.Context,
52	multipartGenerateImageMsg *model.MultipartGenerateImageMsg,
53	) error
54	StartReindexReporting(c context.Context, device string) error
55	StartReindexReportingDeployment(c context.Context, device, deployment, id string) error
56	}
57
58	// NewClient returns a new workflows client
59	func NewClient() Client {	23✔
60	workflowsBaseURL := config.Config.GetString(dconfig.SettingWorkflows)	23✔
61	return &client{	23✔
62	baseURL: workflowsBaseURL,	23✔
63	httpClient: &http.Client{Timeout: defaultTimeout},	23✔
64	}	23✔
65	}	23✔
66
67	type client struct {
68	baseURL string
69	httpClient *http.Client
70	}
71
72	func (c *client) CheckHealth(ctx context.Context) error {	8✔
73	var (	8✔
74	apiErr rest_utils.ApiError	8✔
75	client http.Client	8✔
76	)	8✔
77		8✔
78	if ctx == nil {	10✔
79	ctx = context.Background()	2✔
80	}	2✔
81	if _, ok := ctx.Deadline(); !ok {	12✔
82	var cancel context.CancelFunc	4✔
83	ctx, cancel = context.WithTimeout(ctx, defaultTimeout)	4✔
84	defer cancel()	4✔
85	}	4✔
86	req, _ := http.NewRequestWithContext(	8✔
87	ctx, "GET", c.baseURL+healthURL, nil,	8✔
88	)	8✔
89		8✔
90	rsp, err := client.Do(req)	8✔
91	if err != nil {	10✔
92	return err	2✔
93	}	2✔
94	defer rsp.Body.Close()	6✔
95	if rsp.StatusCode >= http.StatusOK && rsp.StatusCode < 300 {	8✔
96	return nil	2✔
97	}	2✔
98	decoder := json.NewDecoder(rsp.Body)	4✔
99	err = decoder.Decode(&apiErr)	4✔
100	if err != nil {	6✔
101	return errors.Errorf("health check HTTP error: %s", rsp.Status)	2✔
102	}	2✔
103	return &apiErr	2✔
104	}
105
106	func (c *client) StartGenerateArtifact(
107	ctx context.Context,
108	multipartGenerateImageMsg *model.MultipartGenerateImageMsg,
109	) error {	5✔
110	l := log.FromContext(ctx)	5✔
111	l.Debugf("Submit generate artifact: tenantID=%s, artifactID=%s",	5✔
112	multipartGenerateImageMsg.TenantID, multipartGenerateImageMsg.ArtifactID)	5✔
113		5✔
114	workflowsURL := c.baseURL + generateArtifactURL	5✔
115		5✔
116	payload, _ := json.Marshal(multipartGenerateImageMsg)	5✔
117	req, err := http.NewRequest("POST", workflowsURL, strings.NewReader(string(payload)))	5✔
118	if err != nil {	5✔
119	return err	×
120	}	×
121
122	res, err := c.httpClient.Do(req)	5✔
123	if err != nil {	5✔
124	return errors.Wrapf(err, "failed to start workflow: generate_artifact")	×
125	}	×
126	defer res.Body.Close()	5✔
127	if res.StatusCode != http.StatusCreated {	7✔
128	body, err := io.ReadAll(res.Body)	2✔
129	if err != nil {	2✔
130	body = []byte("<failed to read>")	×
131	}	×
132	l.Errorf("generate artifact failed with status %v, response text: %s",	2✔
133	res.StatusCode, body)	2✔
134	return errors.New("failed to start workflow: generate_artifact")	2✔
135	}
136	return nil	3✔
137	}
138
139	func (c *client) StartReindexReporting(ctx context.Context, device string) error {	9✔
140	if _, ok := ctx.Deadline(); !ok {	18✔
141	var cancel context.CancelFunc	9✔
142	ctx, cancel = context.WithTimeout(ctx, defaultTimeout)	9✔
143	defer cancel()	9✔
144	}	9✔
145	id := identity.FromContext(ctx)	9✔
146	if id == nil {	11✔
147	return errors.New("workflows: context lacking tenant identity")	2✔
148	}	2✔
149	wflow := ReindexWorkflow{	7✔
150	RequestID: requestid.FromContext(ctx),	7✔
151	TenantID: id.Tenant,	7✔
152	DeviceID: device,	7✔
153	Service: ServiceDeployments,	7✔
154	}	7✔
155	payload, _ := json.Marshal(wflow)	7✔
156	req, err := http.NewRequestWithContext(ctx,	7✔
157	"POST",	7✔
158	c.baseURL+reindexReportingURL,	7✔
159	bytes.NewReader(payload),	7✔
160	)	7✔
161	if err != nil {	7✔
162	return errors.Wrap(err, "workflows: error preparing HTTP request")	×
163	}	×
164
165	req.Header.Set("Content-Type", "application/json")	7✔
166		7✔
167	rsp, err := c.httpClient.Do(req)	7✔
168	if err != nil {	7✔
169	return errors.Wrap(err, "workflows: failed to trigger reporting reindex")	×
170	}	×
171	defer rsp.Body.Close()	7✔
172		7✔
173	if rsp.StatusCode < 300 {	10✔
174	return nil	3✔
175	}	3✔
176
177	if rsp.StatusCode == http.StatusNotFound {	6✔
178	workflowURIparts := strings.Split(reindexReportingURL, "/")	2✔
179	workflowName := workflowURIparts[len(workflowURIparts)-1]	2✔
180	return errors.New(`workflows: workflow "` + workflowName + `" not defined`)	2✔
181	}	2✔
182
183	return errors.Errorf(	2✔
184	"workflows: unexpected HTTP status from workflows service: %s",	2✔
185	rsp.Status,	2✔
186	)	2✔
187	}
188
189	func (c *client) StartReindexReportingDeployment(ctx context.Context,
190	device, deployment, id string) error {	9✔
191	if _, ok := ctx.Deadline(); !ok {	18✔
192	var cancel context.CancelFunc	9✔
193	ctx, cancel = context.WithTimeout(ctx, defaultTimeout)	9✔
194	defer cancel()	9✔
195	}	9✔
196	ident := identity.FromContext(ctx)	9✔
197	if ident == nil {	11✔
198	return errors.New("workflows: context lacking tenant identity")	2✔
199	}	2✔
200	wflow := ReindexDeploymentWorkflow{	7✔
201	RequestID: requestid.FromContext(ctx),	7✔
202	TenantID: ident.Tenant,	7✔
203	DeviceID: device,	7✔
204	DeploymentID: deployment,	7✔
205	ID: id,	7✔
206	Service: ServiceDeployments,	7✔
207	}	7✔
208	payload, _ := json.Marshal(wflow)	7✔
209	req, err := http.NewRequestWithContext(ctx,	7✔
210	"POST",	7✔
211	c.baseURL+reindexReportingDeploymentURL,	7✔
212	bytes.NewReader(payload),	7✔
213	)	7✔
214	if err != nil {	7✔
NEW 215	return errors.Wrap(err, "workflows: error preparing HTTP request")	×
NEW 216	}	×
217
218	req.Header.Set("Content-Type", "application/json")	7✔
219		7✔
220	rsp, err := c.httpClient.Do(req)	7✔
221	if err != nil {	7✔
NEW 222	return errors.Wrap(err, "workflows: failed to trigger reporting reindex deployment")	×
NEW 223	}	×
224	defer rsp.Body.Close()	7✔
225		7✔
226	if rsp.StatusCode < 300 {	10✔
227	return nil	3✔
228	}	3✔
229
230	if rsp.StatusCode == http.StatusNotFound {	6✔
231	workflowURIparts := strings.Split(reindexReportingDeploymentURL, "/")	2✔
232	workflowName := workflowURIparts[len(workflowURIparts)-1]	2✔
233	return errors.New(`workflows: workflow "` + workflowName + `" not defined`)	2✔
234	}	2✔
235
236	return errors.Errorf(	2✔
237	"workflows: unexpected HTTP status from workflows service: %s",	2✔
238	rsp.Status,	2✔
239	)	2✔
240	}

mendersoftware / deployments / 742310381

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