1284435895

Committed 09 May 2024 10:28AM UTC coverage: 87.639% (+0.09%) from 87.551%

Build # 1284435895

Build Type

Pull #283

gitlab-ci

Committed by

tranchitella

Commit Message

feat: asynchronous processing of webhook requests and timeout

Changelog: process webhook requests asynchronously, returing `202 Accepted` instead of `204 No Content` or `200 OK`
Changelog: add a timeout for webhook requests, defaults to 10 seconds; you can modify it using the `webhooks_timeout_seconds` configuration setting

Ticket: MEN-7227

Signed-off-by: Fabio Tranchitella <fabio.tranchitella@northern.tech>

Pull Request Pull Request #283: feat: asynchronous processing of webhook requests and timeout

Run Details

49 of 53 new or added lines in 6 files covered. (92.45%)

902 existing lines in 13 files now uncovered.

3226 of 3681 relevant lines covered (87.64%)

11.44 hits per line

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

94.64

/app/app_iot_core.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 app

import (
        "context"

        "github.com/mendersoftware/go-lib-micro/log"
        "github.com/pkg/errors"

        "github.com/mendersoftware/iot-manager/client/iotcore"
        "github.com/mendersoftware/iot-manager/model"
)

func assertAWSIntegration(integration model.Integration) error {
        if err := integration.Validate(); err != nil {
                return ErrNoCredentials
        } else if integration.Credentials.Type != model.CredentialTypeAWS {
                return ErrNoCredentials
        }
        return nil
}

func (a *app) provisionIoTCoreDevice(
        ctx context.Context,
        deviceID string,
        integration model.Integration,
        device *iotcore.Device,
) error {
        if err := assertAWSIntegration(integration); err != nil {
                return err
        }

        dev, err := a.iotcoreClient.UpsertDevice(ctx,
                *integration.Credentials.AWSCredentials,
                deviceID,
                device,
                *integration.Credentials.AWSCredentials.DevicePolicyName)
        if err != nil {
                return errors.Wrap(err, "failed to update iotcore devices")
        }

        err = a.deployConfiguration(ctx, deviceID, dev)
        return err
}

func (a *app) setDeviceStatusIoTCore(ctx context.Context, deviceID string, status model.Status,
        integration model.Integration) error {
        if err := assertAWSIntegration(integration); err != nil {
                return err
        }
        _, err := a.iotcoreClient.UpsertDevice(
                ctx,
                *integration.Credentials.AWSCredentials,
                deviceID,
                &iotcore.Device{
                        Status: iotcore.NewStatusFromMenderStatus(status),
                },
                *integration.Credentials.AWSCredentials.DevicePolicyName,
        )
        return err

}

func (a *app) deployConfiguration(ctx context.Context, deviceID string, dev *iotcore.Device) error {
        if dev.Certificate != "" && dev.PrivateKey != "" && *dev.Endpoint != "" {
                err := a.wf.ProvisionExternalDevice(ctx, deviceID, map[string]string{
                        confKeyAWSCertificate: dev.Certificate,
                        confKeyAWSPrivateKey:  dev.PrivateKey,
                        confKeyAWSEndpoint:    *dev.Endpoint,
                })
                if err != nil {
                        return errors.Wrap(err, "failed to submit iotcore credentials to deviceconfig")
                }
        }
        return nil
}

func (a *app) decommissionIoTCoreDevice(ctx context.Context, deviceID string,
        integration model.Integration) error {
        if err := assertAWSIntegration(integration); err != nil {
                return err
        }
        err := a.iotcoreClient.DeleteDevice(ctx, *integration.Credentials.AWSCredentials, deviceID)
        if err != nil && err != iotcore.ErrDeviceNotFound {
                return errors.Wrap(err, "failed to delete IoT Core device")
        }
        return nil
}

func (a *app) syncIoTCoreDevices(
        ctx context.Context,
        deviceIDs []string,
        integration model.Integration,
        failEarly bool,
) error {
        if err := assertAWSIntegration(integration); err != nil {
                return err
        }
        l := log.FromContext(ctx)

        // Get device authentication
        devAuths, err := a.devauth.GetDevices(ctx, deviceIDs)
        if err != nil {
                return errors.Wrap(err, "app: failed to lookup device authentication")
        }

        statuses := make(map[string]model.Status, len(deviceIDs))
        for _, auth := range devAuths {
                statuses[auth.ID] = auth.Status
        }

        // Find devices that shouldn't exist
        var (
                i int
                j int = len(deviceIDs)
        )
        for i < j {
                id := deviceIDs[i]
                if _, ok := statuses[id]; !ok {
                        l.Warnf("Device '%s' does not have an auth set: deleting device", id)
                        err := a.decommissionDevice(ctx, id)
                        if err != nil && !errors.Is(err, ErrDeviceNotFound) {
                                err = errors.Wrap(err, "app: failed to decommission device")
                                if failEarly {
                                        return err
                                }
                                l.Error(err)
                        }
                        // swap(deviceIDs[i], deviceIDs[j])
                        j--
                        tmp := deviceIDs[i]
                        deviceIDs[i] = deviceIDs[j]
                        deviceIDs[j] = tmp
                        deviceIDs = deviceIDs[:j]
                } else {
                        i++
                }
        }
        for _, deviceID := range deviceIDs {
                // Check if device exists in IoT Core
                dev, err := a.iotcoreClient.GetDevice(
                        ctx,
                        *integration.Credentials.AWSCredentials,
                        deviceID,
                )
                status, ok := statuses[deviceID]
                if err == iotcore.ErrDeviceNotFound {
                        if ok {
                                // Device should exist, let's provision the device.
                                err := a.provisionIoTCoreDevice(ctx, deviceID, integration, &iotcore.Device{
                                        Status: iotcore.NewStatusFromMenderStatus(status),
                                })
                                if err != nil {
                                        err = errors.Wrap(err, "failed to provision missing device")
                                        if failEarly {
                                                return err
                                        }
                                        l.Warn(err)
                                }
                        }
                } else if err != nil {
                        err = errors.Wrap(err, "app: failed to get Thing from IoT Core")
                        if failEarly {
                                return err
                        }
                        l.Warn(err)

                } else if dev.Status != iotcore.NewStatusFromMenderStatus(status) {
                        // Upsert device
                        err := a.setDeviceStatusIoTCore(ctx, dev.ID, status, integration)
                        if err != nil {
                                err = errors.Wrap(err, "failed to update device status")
                                if failEarly {
                                        return err
                                }
                                l.Warn(err)
                        }
                }
        }

        return nil
}

func (a *app) GetDeviceStateIoTCore(
        ctx context.Context,
        deviceID string,
        integration *model.Integration,
) (*model.DeviceState, error) {
        if err := assertAWSIntegration(*integration); err != nil {
                return nil, err
        }
        shadow, err := a.iotcoreClient.GetDeviceShadow(
                ctx,
                *integration.Credentials.AWSCredentials,
                deviceID,
        )
        if err != nil {
                if err == iotcore.ErrDeviceNotFound {
                        return nil, nil
                } else {
                        return nil, errors.Wrap(err, "failed to get the device shadow")
                }
        }
        return &shadow.Payload, nil
}

func (a *app) SetDeviceStateIoTCore(
        ctx context.Context,
        deviceID string,
        integration *model.Integration,
        state *model.DeviceState,
) (*model.DeviceState, error) {
        if state == nil {
                return nil, nil
        }
        if err := assertAWSIntegration(*integration); err != nil {
                return nil, err
        }
        shadow, err := a.iotcoreClient.UpdateDeviceShadow(
                ctx,
                *integration.Credentials.AWSCredentials,
                deviceID,
                iotcore.DeviceShadowUpdate{
                        State: iotcore.DesiredState{
                                Desired: state.Desired,
                        },
                },
        )
        if err != nil {
                if err == iotcore.ErrDeviceNotFound {
                        return nil, nil
                }
                return nil, err
        }
        return &shadow.Payload, 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 app
16
17	import (
18	"context"
19
20	"github.com/mendersoftware/go-lib-micro/log"
21	"github.com/pkg/errors"
22
23	"github.com/mendersoftware/iot-manager/client/iotcore"
24	"github.com/mendersoftware/iot-manager/model"
25	)
26
27	func assertAWSIntegration(integration model.Integration) error {	43✔
28	if err := integration.Validate(); err != nil {	48✔
UNCOV 29	return ErrNoCredentials	5✔
30	} else if integration.Credentials.Type != model.CredentialTypeAWS {	43✔
31	return ErrNoCredentials	×
32	}	×
33	return nil	38✔
34	}
35
36	func (a *app) provisionIoTCoreDevice(
37	ctx context.Context,
38	deviceID string,
39	integration model.Integration,
40	device *iotcore.Device,
UNCOV 41	) error {	6✔
UNCOV 42	if err := assertAWSIntegration(integration); err != nil {	7✔
UNCOV 43	return err	1✔
UNCOV 44	}	1✔
45
UNCOV 46	dev, err := a.iotcoreClient.UpsertDevice(ctx,	5✔
UNCOV 47	*integration.Credentials.AWSCredentials,	5✔
UNCOV 48	deviceID,	5✔
UNCOV 49	device,	5✔
UNCOV 50	*integration.Credentials.AWSCredentials.DevicePolicyName)	5✔
UNCOV 51	if err != nil {	7✔
UNCOV 52	return errors.Wrap(err, "failed to update iotcore devices")	2✔
UNCOV 53	}	2✔
54
UNCOV 55	err = a.deployConfiguration(ctx, deviceID, dev)	3✔
UNCOV 56	return err	3✔
57	}
58
59	func (a *app) setDeviceStatusIoTCore(ctx context.Context, deviceID string, status model.Status,
60	integration model.Integration) error {	9✔
61	if err := assertAWSIntegration(integration); err != nil {	10✔
UNCOV 62	return err	1✔
UNCOV 63	}	1✔
64	_, err := a.iotcoreClient.UpsertDevice(	8✔
65	ctx,	8✔
66	*integration.Credentials.AWSCredentials,	8✔
67	deviceID,	8✔
68	&iotcore.Device{	8✔
69	Status: iotcore.NewStatusFromMenderStatus(status),	8✔
70	},	8✔
71	*integration.Credentials.AWSCredentials.DevicePolicyName,	8✔
72	)	8✔
73	return err	8✔
74
75	}
76
UNCOV 77	func (a app) deployConfiguration(ctx context.Context, deviceID string, dev iotcore.Device) error {	3✔
UNCOV 78	if dev.Certificate != "" && dev.PrivateKey != "" && *dev.Endpoint != "" {	6✔
UNCOV 79	err := a.wf.ProvisionExternalDevice(ctx, deviceID, map[string]string{	3✔
UNCOV 80	confKeyAWSCertificate: dev.Certificate,	3✔
UNCOV 81	confKeyAWSPrivateKey: dev.PrivateKey,	3✔
UNCOV 82	confKeyAWSEndpoint: *dev.Endpoint,	3✔
UNCOV 83	})	3✔
UNCOV 84	if err != nil {	4✔
UNCOV 85	return errors.Wrap(err, "failed to submit iotcore credentials to deviceconfig")	1✔
UNCOV 86	}	1✔
87	}
UNCOV 88	return nil	2✔
89	}
90
91	func (a *app) decommissionIoTCoreDevice(ctx context.Context, deviceID string,
92	integration model.Integration) error {	10✔
93	if err := assertAWSIntegration(integration); err != nil {	11✔
UNCOV 94	return err	1✔
UNCOV 95	}	1✔
96	err := a.iotcoreClient.DeleteDevice(ctx, *integration.Credentials.AWSCredentials, deviceID)	9✔
97	if err != nil && err != iotcore.ErrDeviceNotFound {	14✔
98	return errors.Wrap(err, "failed to delete IoT Core device")	5✔
99	}	5✔
UNCOV 100	return nil	4✔
101	}
102
103	func (a *app) syncIoTCoreDevices(
104	ctx context.Context,
105	deviceIDs []string,
106	integration model.Integration,
107	failEarly bool,
108	) error {	11✔
109	if err := assertAWSIntegration(integration); err != nil {	12✔
UNCOV 110	return err	1✔
UNCOV 111	}	1✔
112	l := log.FromContext(ctx)	10✔
113		10✔
114	// Get device authentication	10✔
115	devAuths, err := a.devauth.GetDevices(ctx, deviceIDs)	10✔
116	if err != nil {	11✔
UNCOV 117	return errors.Wrap(err, "app: failed to lookup device authentication")	1✔
UNCOV 118	}	1✔
119
120	statuses := make(map[string]model.Status, len(deviceIDs))	9✔
121	for _, auth := range devAuths {	48✔
122	statuses[auth.ID] = auth.Status	39✔
123	}	39✔
124
125	// Find devices that shouldn't exist
126	var (	9✔
127	i int	9✔
128	j int = len(deviceIDs)	9✔
129	)	9✔
130	for i < j {	54✔
131	id := deviceIDs[i]	45✔
132	if _, ok := statuses[id]; !ok {	51✔
133	l.Warnf("Device '%s' does not have an auth set: deleting device", id)	6✔
134	err := a.decommissionDevice(ctx, id)	6✔
135	if err != nil && !errors.Is(err, ErrDeviceNotFound) {	7✔
UNCOV 136	err = errors.Wrap(err, "app: failed to decommission device")	1✔
UNCOV 137	if failEarly {	2✔
UNCOV 138	return err	1✔
UNCOV 139	}	1✔
140	l.Error(err)	×
141	}
142	// swap(deviceIDs[i], deviceIDs[j])
143	j--	5✔
144	tmp := deviceIDs[i]	5✔
145	deviceIDs[i] = deviceIDs[j]	5✔
146	deviceIDs[j] = tmp	5✔
147	deviceIDs = deviceIDs[:j]	5✔
148	} else {	39✔
149	i++	39✔
150	}	39✔
151	}
152	for _, deviceID := range deviceIDs {	47✔
153	// Check if device exists in IoT Core	39✔
154	dev, err := a.iotcoreClient.GetDevice(	39✔
155	ctx,	39✔
156	*integration.Credentials.AWSCredentials,	39✔
157	deviceID,	39✔
158	)	39✔
159	status, ok := statuses[deviceID]	39✔
160	if err == iotcore.ErrDeviceNotFound {	41✔
UNCOV 161	if ok {	4✔
UNCOV 162	// Device should exist, let's provision the device.	2✔
UNCOV 163	err := a.provisionIoTCoreDevice(ctx, deviceID, integration, &iotcore.Device{	2✔
UNCOV 164	Status: iotcore.NewStatusFromMenderStatus(status),	2✔
UNCOV 165	})	2✔
UNCOV 166	if err != nil {	3✔
UNCOV 167	err = errors.Wrap(err, "failed to provision missing device")	1✔
UNCOV 168	if failEarly {	2✔
UNCOV 169	return err	1✔
UNCOV 170	}	1✔
171	l.Warn(err)	×
172	}
173	}
174	} else if err != nil {	39✔
175	err = errors.Wrap(err, "app: failed to get Thing from IoT Core")	2✔
176	if failEarly {	3✔
UNCOV 177	return err	1✔
UNCOV 178	}	1✔
179	l.Warn(err)	1✔
180
181	} else if dev.Status != iotcore.NewStatusFromMenderStatus(status) {	40✔
182	// Upsert device	5✔
183	err := a.setDeviceStatusIoTCore(ctx, dev.ID, status, integration)	5✔
184	if err != nil {	6✔
UNCOV 185	err = errors.Wrap(err, "failed to update device status")	1✔
UNCOV 186	if failEarly {	2✔
UNCOV 187	return err	1✔
UNCOV 188	}	1✔
189	l.Warn(err)	×
190	}
191	}
192	}
193
194	return nil	5✔
195	}
196
197	func (a *app) GetDeviceStateIoTCore(
198	ctx context.Context,
199	deviceID string,
200	integration *model.Integration,
UNCOV 201	) (*model.DeviceState, error) {	4✔
UNCOV 202	if err := assertAWSIntegration(*integration); err != nil {	5✔
UNCOV 203	return nil, err	1✔
UNCOV 204	}	1✔
UNCOV 205	shadow, err := a.iotcoreClient.GetDeviceShadow(	3✔
UNCOV 206	ctx,	3✔
UNCOV 207	*integration.Credentials.AWSCredentials,	3✔
UNCOV 208	deviceID,	3✔
UNCOV 209	)	3✔
UNCOV 210	if err != nil {	5✔
UNCOV 211	if err == iotcore.ErrDeviceNotFound {	3✔
UNCOV 212	return nil, nil	1✔
UNCOV 213	} else {	2✔
UNCOV 214	return nil, errors.Wrap(err, "failed to get the device shadow")	1✔
UNCOV 215	}	1✔
216	}
UNCOV 217	return &shadow.Payload, nil	1✔
218	}
219
220	func (a *app) SetDeviceStateIoTCore(
221	ctx context.Context,
222	deviceID string,
223	integration *model.Integration,
224	state *model.DeviceState,
UNCOV 225	) (*model.DeviceState, error) {	3✔
UNCOV 226	if state == nil {	3✔
227	return nil, nil	×
228	}	×
UNCOV 229	if err := assertAWSIntegration(*integration); err != nil {	3✔
230	return nil, err	×
231	}	×
UNCOV 232	shadow, err := a.iotcoreClient.UpdateDeviceShadow(	3✔
UNCOV 233	ctx,	3✔
UNCOV 234	*integration.Credentials.AWSCredentials,	3✔
UNCOV 235	deviceID,	3✔
UNCOV 236	iotcore.DeviceShadowUpdate{	3✔
UNCOV 237	State: iotcore.DesiredState{	3✔
UNCOV 238	Desired: state.Desired,	3✔
UNCOV 239	},	3✔
UNCOV 240	},	3✔
UNCOV 241	)	3✔
UNCOV 242	if err != nil {	5✔
UNCOV 243	if err == iotcore.ErrDeviceNotFound {	3✔
UNCOV 244	return nil, nil	1✔
UNCOV 245	}	1✔
UNCOV 246	return nil, err	1✔
247	}
UNCOV 248	return &shadow.Payload, nil	1✔
249	}

mendersoftware / iot-manager / 1284435895

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