Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

chore(pkger): handle edge cases that can manifest from users in the platform and stateful pkg applications 

references: #17434
pull/17981/head
Johnny Steenbergen 2020-05-06 11:09:16 -07:00 committed by Johnny Steenbergen
parent 9ab4447617
commit f575ea8d41
3 changed files with 106 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
cmd/influxd/launcher/pkger_test.go
View File

@ -7,6 +7,7 @@ import (
"io/ioutil"
nethttp "net/http"
"net/http/httptest"
"sort"
"testing"
"time"
@ -591,7 +592,7 @@ func TestLauncher_Pkger(t *testing.T) {
})
})
t.Run("apply a pkg with a stack", func(t *testing.T) {
t.Run("apply a pkg with a stack and all resources", func(t *testing.T) {
testStackApplyFn := func(t *testing.T) (pkger.Summary, pkger.Stack, func()) {
t.Helper()
@ -1038,6 +1039,55 @@ func TestLauncher_Pkger(t *testing.T) {
}
})
})
t.Run("apply should handle cases where users have changed platform data", func(t *testing.T) {
t.Run("when a user has deleted a variable that was previously created by a stack", func(t *testing.T) {
testUserDeletedVariable := func(t *testing.T, actionFn func(t *testing.T, stackID influxdb.ID, initialVarObj pkger.Object, initialSum pkger.Summary)) {
t.Helper()
stack, cleanup := newStackFn(t, pkger.Stack{})
defer cleanup()
varObj := newVariableObject("var-1", "", "")
pkg := newPkg(varObj)
initialSum, _, err := svc.Apply(ctx, l.Org.ID, l.User.ID, pkg, pkger.ApplyWithStackID(stack.ID))
require.NoError(t, err)
require.Len(t, initialSum.Variables, 1)
require.NotZero(t, initialSum.Variables[0].ID)
resourceCheck.mustDeleteVariable(t, influxdb.ID(initialSum.Variables[0].ID))
actionFn(t, stack.ID, varObj, initialSum)
}
t.Run("should create new resource when attempting to update", func(t *testing.T) {
testUserDeletedVariable(t, func(t *testing.T, stackID influxdb.ID, initialVarObj pkger.Object, initialSum pkger.Summary) {
pkg := newPkg(initialVarObj)
updateSum, _, err := svc.Apply(ctx, l.Org.ID, l.User.ID, pkg, pkger.ApplyWithStackID(stackID))
require.NoError(t, err)
require.Len(t, updateSum.Variables, 1)
initVar, updateVar := initialSum.Variables[0], updateSum.Variables[0]
assert.NotEqual(t, initVar.ID, updateVar.ID)
initVar.ID, updateVar.ID = 0, 0
assert.Equal(t, initVar, updateVar)
})
})
t.Run("should not error when attempting to remove", func(t *testing.T) {
testUserDeletedVariable(t, func(t *testing.T, stackID influxdb.ID, initialVarObj pkger.Object, initialSum pkger.Summary) {
_, _, err := svc.Apply(
ctx,
l.Org.ID,
l.User.ID,
newPkg( /* empty stack to remove prev variable */ ),
pkger.ApplyWithStackID(stackID),
)
require.NoError(t, err)
})
})
})
})
})
t.Run("errors incurred during application of package rolls back to state before package", func(t *testing.T) {
@ -1386,6 +1436,7 @@ spec:
labels := sum.Labels
require.Len(t, labels, 2)
sortLabels(labels)
assert.Equal(t, "label-1", labels[0].Name)
assert.Equal(t, "the 2nd label", labels[1].Name)
@ -1397,6 +1448,7 @@ spec:
checks := sum.Checks
require.Len(t, checks, 2)
sortChecks(checks)
assert.Equal(t, "check 0 name", checks[0].Check.GetName())
hasLabelAssociations(t, checks[0].LabelAssociations, 1, "label-1")
assert.Equal(t, "check-1", checks[1].Check.GetName())
@ -1691,6 +1743,7 @@ spec:
labels := newSum.Labels
require.Len(t, labels, 2)
sortLabels(labels)
assert.Zero(t, labels[0].ID)
assert.Equal(t, "label-1", labels[0].Name)
assert.Zero(t, labels[1].ID)
@ -1704,6 +1757,7 @@ spec:
checks := newSum.Checks
require.Len(t, checks, 2)
sortChecks(checks)
assert.Equal(t, "check 0 name", checks[0].Check.GetName())
hasLabelAssociations(t, checks[0].LabelAssociations, 1, "label-1")
assert.Equal(t, "check-1", checks[1].Check.GetName())
@ -2774,3 +2828,22 @@ func (r resourceChecker) mustGetVariable(t *testing.T, getOpt getResourceOptFn)
require.NoError(t, err)
return l
}
func (r resourceChecker) mustDeleteVariable(t *testing.T, id influxdb.ID) {
t.Helper()
err := r.tl.VariableService(t).DeleteVariable(ctx, id)
require.NoError(t, err)
}
func sortChecks(checks []pkger.SummaryCheck) {
sort.Slice(checks, func(i, j int) bool {
return checks[i].Check.GetName() < checks[j].Check.GetName()
})
}
func sortLabels(labels []pkger.SummaryLabel) {
sort.Slice(labels, func(i, j int) bool {
return labels[i].Name < labels[j].Name
})
}

37
pkger/service.go
View File

@ -957,11 +957,9 @@ func (s *Service) dryRunVariables(ctx context.Context, orgID influxdb.ID, vars m
}
for _, v := range vars {
var existing *influxdb.Variable
existing := mNames[v.parserVar.Name()]
if v.ID() != 0 {
existing = mIDs[v.ID()]
} else {
existing = mNames[v.parserVar.Name()]
}
if IsNew(v.stateStatus) && existing != nil {
v.stateStatus = StateStatusExists
@ -1095,8 +1093,9 @@ func (s *Service) dryRunResourceLabelMapping(ctx context.Context, state *stateCo
ResourceID: ident.id,
ResourceType: ident.resourceType,
})
if err != nil {
return nil, err
if err != nil && influxdb.ErrorCode(err) != influxdb.ENotFound {
msgFmt := fmt.Sprintf("failed to find labels mappings for %s resource[%q]", ident.resourceType, ident.id)
return nil, ierrors.Wrap(err, msgFmt)
}
pkgLabels := labelSlcToMap(pkgResourceLabels)
@ -1235,7 +1234,7 @@ func (s *Service) Apply(ctx context.Context, orgID, userID influxdb.ID, pkg *Pkg
func (s *Service) applyState(ctx context.Context, coordinator *rollbackCoordinator, orgID, userID influxdb.ID, state *stateCoordinator, missingSecrets map[string]string) (e error) {
endpointApp, ruleApp, err := s.applyNotificationGenerator(ctx, userID, state.rules(), state.endpoints())
if err != nil {
return err
return ierrors.Wrap(err, "failed to setup notification generator")
}
// each grouping here runs for its entirety, then returns an error that
@ -2395,9 +2394,9 @@ func (s *Service) rollbackVariables(ctx context.Context, variables []*stateVaria
err = ierrors.Wrap(s.varSVC.CreateVariable(ctx, v.existing), "rolling back removed variable")
case StateStatusExists:
_, err = s.varSVC.UpdateVariable(ctx, v.ID(), &influxdb.VariableUpdate{
Name: v.parserVar.Name(),
Description: v.parserVar.Description,
Arguments: v.parserVar.influxVarArgs(),
Name: v.existing.Name,
Description: v.existing.Description,
Arguments: v.existing.Arguments,
})
err = ierrors.Wrap(err, "rolling back updated variable")
default:
@ -2421,23 +2420,28 @@ func (s *Service) rollbackVariables(ctx context.Context, variables []*stateVaria
}
func (s *Service) applyVariable(ctx context.Context, v *stateVariable) (influxdb.Variable, error) {
switch v.stateStatus {
case StateStatusRemove:
if err := s.varSVC.DeleteVariable(ctx, v.id); err != nil {
return influxdb.Variable{}, err
switch {
case IsRemoval(v.stateStatus):
if err := s.varSVC.DeleteVariable(ctx, v.id); err != nil && influxdb.ErrorCode(err) != influxdb.ENotFound {
return influxdb.Variable{}, ierrors.Wrap(err, "removing existing variable")
}
if v.existing == nil {
return influxdb.Variable{}, nil
}
return *v.existing, nil
case StateStatusExists:
case IsExisting(v.stateStatus) && v.existing != nil:
updatedVar, err := s.varSVC.UpdateVariable(ctx, v.ID(), &influxdb.VariableUpdate{
Name: v.parserVar.Name(),
Description: v.parserVar.Description,
Arguments: v.parserVar.influxVarArgs(),
})
if err != nil {
return influxdb.Variable{}, err
return influxdb.Variable{}, ierrors.Wrap(err, "updating existing variable")
}
return *updatedVar, nil
default:
// when an existing variable (referenced in stack) has been deleted by a user
// then the resource is created anew to get it back to the expected state.
influxVar := influxdb.Variable{
OrganizationID: v.orgID,
Name: v.parserVar.Name(),
@ -2446,9 +2450,8 @@ func (s *Service) applyVariable(ctx context.Context, v *stateVariable) (influxdb
}
err := s.varSVC.CreateVariable(ctx, &influxVar)
if err != nil {
return influxdb.Variable{}, err
return influxdb.Variable{}, ierrors.Wrap(err, "creating new variable")
}
return influxVar, nil
}
}

12
pkger/service_test.go
View File

@ -7,6 +7,7 @@ import (
"fmt"
"math/rand"
"regexp"
"sort"
"strconv"
"testing"
"time"
@ -1482,6 +1483,12 @@ func TestService(t *testing.T) {
}
}
sortLabelsByName := func(labels []SummaryLabel) {
sort.Slice(labels, func(i, j int) bool {
return labels[i].Name < labels[j].Name
})
}
t.Run("with existing resources", func(t *testing.T) {
encodeAndDecode := func(t *testing.T, pkg *Pkg) *Pkg {
t.Helper()
@ -2853,9 +2860,13 @@ func TestService(t *testing.T) {
sum := newPkg.Summary()
bkts := sum.Buckets
sort.Slice(bkts, func(i, j int) bool {
return bkts[i].Name < bkts[j].Name
})
require.Len(t, bkts, 2)
for i, actual := range bkts {
sortLabelsByName(actual.LabelAssociations)
assert.Equal(t, strconv.Itoa((i+1)*10), actual.Name)
require.Len(t, actual.LabelAssociations, 2)
assert.Equal(t, "label_1", actual.LabelAssociations[0].Name)
@ -2863,6 +2874,7 @@ func TestService(t *testing.T) {
}
labels := sum.Labels
sortLabelsByName(labels)
require.Len(t, labels, 2)
assert.Equal(t, "label_1", labels[0].Name)
assert.Equal(t, "label_2", labels[1].Name)