Skip to content

Commit 26dad6c

Browse files
tigrannajaryantigrannajaryan
authored
Implement package syncing (#76)
- This implements "Packages" spec section https://github.com/open-telemetry/opamp-spec/blob/main/specification.md#packages-1 - TODO: increase test coverage to handle many more edge cases. - TODO: add package usage to the example Agent/Server.
1 parent a3fd1a4 commit 26dad6c

14 files changed

+953
-47
lines changed

‎client/clientimpl_test.go

Lines changed: 253 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"errors"
66
"math/rand"
77
"net/http"
8+
"net/http/httptest"
89
"net/url"
910
"sync/atomic"
1011
"testing"
@@ -397,6 +398,7 @@ func TestIncludesDetailsOnReconnect(t *testing.T) {
397398
eventually(t, func() bool { return atomic.LoadInt64(&receivedDetails) == 1 })
398399

399400
// close the Agent connection. expect it to reconnect and send details again.
401+
require.NotNil(t, client.conn)
400402
err := client.conn.Close()
401403
assert.NoError(t, err)
402404

@@ -444,7 +446,7 @@ func TestSetEffectiveConfig(t *testing.T) {
444446
settings.OpAMPServerURL = "ws://" + srv.Endpoint
445447
prepareClient(t, &settings, client)
446448

447-
assert.NoError(t, client.Start(context.Background(), settings))
449+
require.NoError(t, client.Start(context.Background(), settings))
448450

449451
// Verify config is delivered.
450452
eventually(
@@ -972,3 +974,253 @@ func TestRemoteConfigUpdate(t *testing.T) {
972974
})
973975
}
974976
}
977+
978+
type packageTestCase struct {
979+
name string
980+
errorOnCallback bool
981+
available *protobufs.PackagesAvailable
982+
expectedStatus *protobufs.PackageStatuses
983+
expectedFileContent map[string][]byte
984+
}
985+
986+
const packageUpdateErrorMsg = "cannot update packages"
987+
988+
func verifyUpdatePackages(t *testing.T, testCase packageTestCase) {
989+
testClients(t, func(t *testing.T, client OpAMPClient) {
990+
991+
// Start a Server.
992+
srv := internal.StartMockServer(t)
993+
srv.EnableExpectMode()
994+
995+
localPackageState := internal.NewInMemPackagesStore()
996+
997+
var syncerDoneCh <-chan struct{}
998+
999+
// Prepare a callback that returns either success or failure.
1000+
onPackagesAvailable := func(ctx context.Context, packages *protobufs.PackagesAvailable, syncer types.PackagesSyncer) error {
1001+
if testCase.errorOnCallback {
1002+
return errors.New(packageUpdateErrorMsg)
1003+
} else {
1004+
syncerDoneCh = syncer.Done()
1005+
err := syncer.Sync(ctx)
1006+
require.NoError(t, err)
1007+
return nil
1008+
}
1009+
}
1010+
1011+
// Start a client.
1012+
settings := types.StartSettings{
1013+
OpAMPServerURL: "ws://" + srv.Endpoint,
1014+
Callbacks: types.CallbacksStruct{
1015+
OnPackagesAvailableFunc: onPackagesAvailable,
1016+
},
1017+
PackagesStateProvider: localPackageState,
1018+
}
1019+
prepareClient(t, &settings, client)
1020+
1021+
// Client --->
1022+
assert.NoError(t, client.Start(context.Background(), settings))
1023+
1024+
// ---> Server
1025+
srv.Expect(func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
1026+
// Send the packages to the Agent.
1027+
return &protobufs.ServerToAgent{
1028+
InstanceUid: msg.InstanceUid,
1029+
PackagesAvailable: testCase.available,
1030+
}
1031+
})
1032+
1033+
// The Agent will try to install the packages and will send the status
1034+
// report about it back to the Server.
1035+
1036+
var lastStatusHash []byte
1037+
1038+
// ---> Server
1039+
// Wait for the expected package statuses to be received.
1040+
srv.EventuallyExpect("full PackageStatuses",
1041+
func(msg *protobufs.AgentToServer) (*protobufs.ServerToAgent, bool) {
1042+
expectedStatusReceived := false
1043+
1044+
status := msg.PackageStatuses
1045+
require.NotNil(t, status)
1046+
assert.EqualValues(t, testCase.expectedStatus.ServerProvidedAllPackagesHash, status.ServerProvidedAllPackagesHash)
1047+
lastStatusHash = status.Hash
1048+
1049+
// Verify individual package statuses.
1050+
for name, pkgExpected := range testCase.expectedStatus.Packages {
1051+
pkgStatus := status.Packages[name]
1052+
if pkgStatus == nil {
1053+
// Package status not yet included in the report.
1054+
continue
1055+
}
1056+
switch pkgStatus.Status {
1057+
case protobufs.PackageStatus_InstallFailed:
1058+
assert.Contains(t, pkgStatus.ErrorMessage, pkgExpected.ErrorMessage)
1059+
1060+
case protobufs.PackageStatus_Installed:
1061+
assert.EqualValues(t, pkgExpected.AgentHasHash, pkgStatus.AgentHasHash)
1062+
assert.EqualValues(t, pkgExpected.AgentHasVersion, pkgStatus.AgentHasVersion)
1063+
assert.Empty(t, pkgStatus.ErrorMessage)
1064+
default:
1065+
assert.Empty(t, pkgStatus.ErrorMessage)
1066+
}
1067+
assert.EqualValues(t, pkgExpected.ServerOfferedHash, pkgStatus.ServerOfferedHash)
1068+
assert.EqualValues(t, pkgExpected.ServerOfferedVersion, pkgStatus.ServerOfferedVersion)
1069+
1070+
if pkgStatus.Status == pkgExpected.Status {
1071+
expectedStatusReceived = true
1072+
assert.Len(t, status.Packages, len(testCase.available.Packages))
1073+
}
1074+
}
1075+
assert.NotNil(t, status.Hash)
1076+
1077+
return &protobufs.ServerToAgent{InstanceUid: msg.InstanceUid}, expectedStatusReceived
1078+
})
1079+
1080+
if syncerDoneCh != nil {
1081+
// Wait until all syncing is done.
1082+
<-syncerDoneCh
1083+
1084+
for pkgName, receivedContent := range localPackageState.GetContent() {
1085+
expectedContent := testCase.expectedFileContent[pkgName]
1086+
assert.EqualValues(t, expectedContent, receivedContent)
1087+
}
1088+
}
1089+
1090+
// Client --->
1091+
// Trigger another status report by setting AgentDescription.
1092+
_ = client.SetAgentDescription(client.AgentDescription())
1093+
1094+
// ---> Server
1095+
srv.EventuallyExpect("compressed PackageStatuses",
1096+
func(msg *protobufs.AgentToServer) (*protobufs.ServerToAgent, bool) {
1097+
// Ensure that compressed status is received.
1098+
status := msg.PackageStatuses
1099+
require.NotNil(t, status)
1100+
compressedReceived := status.ServerProvidedAllPackagesHash == nil
1101+
if compressedReceived {
1102+
assert.Nil(t, status.ServerProvidedAllPackagesHash)
1103+
assert.Nil(t, status.Packages)
1104+
}
1105+
assert.NotNil(t, status.Hash)
1106+
assert.Equal(t, lastStatusHash, status.Hash)
1107+
1108+
response := &protobufs.ServerToAgent{InstanceUid: msg.InstanceUid}
1109+
1110+
if compressedReceived {
1111+
// Ask for full report again.
1112+
response.Flags = protobufs.ServerToAgent_ReportPackageStatuses
1113+
} else {
1114+
// Keep triggering status report by setting AgentDescription
1115+
// until the compressed PackageStatuses arrives.
1116+
_ = client.SetAgentDescription(client.AgentDescription())
1117+
}
1118+
1119+
return response, compressedReceived
1120+
})
1121+
1122+
// Shutdown the Server.
1123+
srv.Close()
1124+
1125+
// Shutdown the client.
1126+
err := client.Stop(context.Background())
1127+
assert.NoError(t, err)
1128+
})
1129+
}
1130+
1131+
// Downloadable package file constants.
1132+
const packageFileURL = "/validfile.pkg"
1133+
1134+
var packageFileContent = []byte("Package File Content")
1135+
1136+
func createDownloadSrv(t *testing.T) *httptest.Server {
1137+
m := http.NewServeMux()
1138+
m.HandleFunc(packageFileURL,
1139+
func(w http.ResponseWriter, r *http.Request) {
1140+
w.WriteHeader(http.StatusOK)
1141+
_, err := w.Write(packageFileContent)
1142+
assert.NoError(t, err)
1143+
},
1144+
)
1145+
1146+
srv := httptest.NewServer(m)
1147+
1148+
u, err := url.Parse(srv.URL)
1149+
if err != nil {
1150+
t.Fatal(err)
1151+
}
1152+
endpoint := u.Host
1153+
testhelpers.WaitForEndpoint(endpoint)
1154+
1155+
return srv
1156+
}
1157+
1158+
func createPackageTestCase(name string, downloadSrv *httptest.Server) packageTestCase {
1159+
return packageTestCase{
1160+
name: name,
1161+
errorOnCallback: false,
1162+
available: &protobufs.PackagesAvailable{
1163+
Packages: map[string]*protobufs.PackageAvailable{
1164+
"package1": {
1165+
Type: protobufs.PackageAvailable_TopLevelPackage,
1166+
Version: "1.0.0",
1167+
File: &protobufs.DownloadableFile{
1168+
DownloadUrl: downloadSrv.URL + packageFileURL,
1169+
ContentHash: []byte{4, 5},
1170+
},
1171+
Hash: []byte{1, 2, 3},
1172+
},
1173+
},
1174+
AllPackagesHash: []byte{1, 2, 3, 4, 5},
1175+
},
1176+
1177+
expectedStatus: &protobufs.PackageStatuses{
1178+
Packages: map[string]*protobufs.PackageStatus{
1179+
"package1": {
1180+
Name: "package1",
1181+
AgentHasVersion: "1.0.0",
1182+
AgentHasHash: []byte{1, 2, 3},
1183+
ServerOfferedVersion: "1.0.0",
1184+
ServerOfferedHash: []byte{1, 2, 3},
1185+
Status: protobufs.PackageStatus_Installed,
1186+
ErrorMessage: "",
1187+
},
1188+
},
1189+
ServerProvidedAllPackagesHash: []byte{1, 2, 3, 4, 5},
1190+
},
1191+
1192+
expectedFileContent: map[string][]byte{
1193+
"package1": packageFileContent,
1194+
},
1195+
}
1196+
}
1197+
1198+
func TestUpdatePackages(t *testing.T) {
1199+
1200+
downloadSrv := createDownloadSrv(t)
1201+
defer downloadSrv.Close()
1202+
1203+
// A success case.
1204+
var tests []packageTestCase
1205+
tests = append(tests, createPackageTestCase("success", downloadSrv))
1206+
1207+
// A case when downloading the file fails because the URL is incorrect.
1208+
notFound := createPackageTestCase("downloadable file not found", downloadSrv)
1209+
notFound.available.Packages["package1"].File.DownloadUrl = downloadSrv.URL + "/notfound"
1210+
notFound.expectedStatus.Packages["package1"].Status = protobufs.PackageStatus_InstallFailed
1211+
notFound.expectedStatus.Packages["package1"].ErrorMessage = "cannot download"
1212+
tests = append(tests, notFound)
1213+
1214+
// A case when OnPackagesAvailable callback returns an error.
1215+
errorOnCallback := createPackageTestCase("error on callback", downloadSrv)
1216+
errorOnCallback.expectedStatus.Packages["package1"].Status = protobufs.PackageStatus_InstallFailed
1217+
errorOnCallback.expectedStatus.Packages["package1"].ErrorMessage = packageUpdateErrorMsg
1218+
errorOnCallback.errorOnCallback = true
1219+
tests = append(tests, errorOnCallback)
1220+
1221+
for _, test := range tests {
1222+
t.Run(test.name, func(t *testing.T) {
1223+
verifyUpdatePackages(t, test)
1224+
})
1225+
}
1226+
}

‎client/httpclient.go

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -76,5 +76,11 @@ func (c *httpClient) runUntilStopped(ctx context.Context) {
7676
// Start the HTTP sender. This will make request/responses with retries for
7777
// failures and will wait with configured polling interval if there is nothing
7878
// to send.
79-
c.sender.Run(ctx, c.opAMPServerURL, c.common.Callbacks, &c.common.ClientSyncedState)
79+
c.sender.Run(
80+
ctx,
81+
c.opAMPServerURL,
82+
c.common.Callbacks,
83+
&c.common.ClientSyncedState,
84+
c.common.PackagesStateProvider,
85+
)
8086
}

‎client/internal/clientcommon.go

Lines changed: 38 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -10,14 +10,15 @@ import (
1010

1111
"github.com/open-telemetry/opamp-go/client/types"
1212
"github.com/open-telemetry/opamp-go/protobufs"
13+
"google.golang.org/protobuf/proto"
1314
)
1415

1516
var (
1617
ErrAgentDescriptionMissing = errors.New("AgentDescription is nil")
1718
ErrAgentDescriptionNoAttributes = errors.New("AgentDescription has no attributes defined")
18-
errRemoteConfigStatusMissing = errors.New("RemoteConfigStatus is not set")
19-
errAlreadyStarted = errors.New("already started")
20-
errCannotStopNotStarted = errors.New("cannot stop because not started")
19+
20+
errAlreadyStarted = errors.New("already started")
21+
errCannotStopNotStarted = errors.New("cannot stop because not started")
2122
)
2223

2324
// ClientCommon contains the OpAMP logic that is common between WebSocket and
@@ -29,6 +30,9 @@ type ClientCommon struct {
2930
// Client state storage. This is needed if the Server asks to report the state.
3031
ClientSyncedState ClientSyncedState
3132

33+
// PackagesStateProvider provides access to the local state of packages.
34+
PackagesStateProvider types.PackagesStateProvider
35+
3236
// The transport-specific sender.
3337
sender Sender
3438

@@ -59,6 +63,7 @@ func (c *ClientCommon) PrepareStart(_ context.Context, settings types.StartSetti
5963
return ErrAgentDescriptionMissing
6064
}
6165

66+
// Prepare remote config status.
6267
if settings.RemoteConfigStatus == nil {
6368
// RemoteConfigStatus is not provided. Start with empty.
6469
settings.RemoteConfigStatus = &protobufs.RemoteConfigStatus{
@@ -70,6 +75,27 @@ func (c *ClientCommon) PrepareStart(_ context.Context, settings types.StartSetti
7075
return err
7176
}
7277

78+
// Prepare package statuses.
79+
c.PackagesStateProvider = settings.PackagesStateProvider
80+
var packageStatuses *protobufs.PackageStatuses
81+
if settings.PackagesStateProvider != nil {
82+
// Set package status from the value previously saved in the PackagesStateProvider.
83+
var err error
84+
packageStatuses, err = settings.PackagesStateProvider.LastReportedStatuses()
85+
if err != nil {
86+
return err
87+
}
88+
}
89+
90+
if packageStatuses == nil {
91+
// PackageStatuses is not provided. Start with empty.
92+
packageStatuses = &protobufs.PackageStatuses{}
93+
}
94+
if err := c.ClientSyncedState.SetPackageStatuses(packageStatuses); err != nil {
95+
return err
96+
}
97+
98+
// Prepare callbacks.
7399
c.Callbacks = settings.Callbacks
74100
if c.Callbacks == nil {
75101
// Make sure it is always safe to call Callbacks.
@@ -154,24 +180,24 @@ func (c *ClientCommon) PrepareFirstMessage(ctx context.Context) error {
154180
msg.StatusReport = &protobufs.StatusReport{}
155181
}
156182
msg.StatusReport.AgentDescription = c.ClientSyncedState.AgentDescription()
157-
158183
msg.StatusReport.EffectiveConfig = cfg
159-
160184
msg.StatusReport.RemoteConfigStatus = c.ClientSyncedState.RemoteConfigStatus()
185+
msg.PackageStatuses = c.ClientSyncedState.PackageStatuses()
161186

162-
if msg.PackageStatuses == nil {
163-
msg.PackageStatuses = &protobufs.PackageStatuses{}
187+
if c.PackagesStateProvider != nil {
188+
// We have a state provider, so package related capabilities can work.
189+
msg.StatusReport.Capabilities |= protobufs.AgentCapabilities_AcceptsPackages
190+
msg.StatusReport.Capabilities |= protobufs.AgentCapabilities_ReportsPackageStatuses
164191
}
165-
166-
// TODO: set PackageStatuses.ServerProvidedAllPackagesHash field and
167-
// handle the Hashes for PackageStatuses properly.
168192
},
169193
)
170194
return nil
171195
}
172196

197+
// AgentDescription returns the current state of the AgentDescription.
173198
func (c *ClientCommon) AgentDescription() *protobufs.AgentDescription {
174-
return c.ClientSyncedState.AgentDescription()
199+
// Return a cloned copy to allow caller to do whatever they want with the result.
200+
return proto.Clone(c.ClientSyncedState.AgentDescription()).(*protobufs.AgentDescription)
175201
}
176202

177203
// SetAgentDescription sends a status update to the Server with the new AgentDescription
@@ -183,7 +209,7 @@ func (c *ClientCommon) SetAgentDescription(descr *protobufs.AgentDescription) er
183209
return err
184210
}
185211
c.sender.NextMessage().UpdateStatus(func(statusReport *protobufs.StatusReport) {
186-
statusReport.AgentDescription = descr
212+
statusReport.AgentDescription = c.ClientSyncedState.AgentDescription()
187213
})
188214
c.sender.ScheduleSend()
189215
return nil

0 commit comments

Comments
 (0)