GitOps for k8s
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

daemon_test.go 22KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915
  1. package daemon
  2. import (
  3. "bufio"
  4. "context"
  5. "fmt"
  6. "os"
  7. "path/filepath"
  8. "strings"
  9. "sync"
  10. "testing"
  11. "time"
  12. "github.com/go-kit/kit/log"
  13. "github.com/stretchr/testify/assert"
  14. "github.com/weaveworks/flux"
  15. "github.com/weaveworks/flux/api/v10"
  16. "github.com/weaveworks/flux/api/v11"
  17. "github.com/weaveworks/flux/api/v6"
  18. "github.com/weaveworks/flux/api/v9"
  19. "github.com/weaveworks/flux/cluster"
  20. "github.com/weaveworks/flux/cluster/kubernetes"
  21. kresource "github.com/weaveworks/flux/cluster/kubernetes/resource"
  22. "github.com/weaveworks/flux/event"
  23. "github.com/weaveworks/flux/git"
  24. "github.com/weaveworks/flux/git/gittest"
  25. "github.com/weaveworks/flux/image"
  26. "github.com/weaveworks/flux/job"
  27. "github.com/weaveworks/flux/policy"
  28. "github.com/weaveworks/flux/registry"
  29. registryMock "github.com/weaveworks/flux/registry/mock"
  30. "github.com/weaveworks/flux/resource"
  31. "github.com/weaveworks/flux/update"
  32. )
  33. const (
  34. // These have to match the values in cluster/kubernetes/testfiles/data.go
  35. wl = "default:deployment/helloworld"
  36. container = "greeter"
  37. ns = "default"
  38. oldHelloImage = "quay.io/weaveworks/helloworld:3" // older in time but newer version!
  39. newHelloImage = "quay.io/weaveworks/helloworld:2"
  40. currentHelloImage = "quay.io/weaveworks/helloworld:master-a000001"
  41. anotherWl = "another:deployment/service"
  42. anotherContainer = "it-doesn't-matter"
  43. anotherImage = "another/service:latest"
  44. invalidNS = "adsajkfldsa"
  45. testVersion = "test"
  46. )
  47. var testBytes = []byte(`{}`)
  48. const timeout = 10 * time.Second
  49. // When I ping, I should get a response
  50. func TestDaemon_Ping(t *testing.T) {
  51. d, start, clean, _, _, _ := mockDaemon(t)
  52. start()
  53. defer clean()
  54. ctx := context.Background()
  55. if d.Ping(ctx) != nil {
  56. t.Fatal("Cluster did not return valid nil ping")
  57. }
  58. }
  59. // When I ask a version, I should get a version
  60. func TestDaemon_Version(t *testing.T) {
  61. d, start, clean, _, _, _ := mockDaemon(t)
  62. start()
  63. defer clean()
  64. ctx := context.Background()
  65. v, err := d.Version(ctx)
  66. if err != nil {
  67. t.Fatalf("Error: %s", err.Error())
  68. }
  69. if v != testVersion {
  70. t.Fatalf("Expected %v but got %v", testVersion, v)
  71. }
  72. }
  73. // When I export it should export the current (mocked) k8s cluster
  74. func TestDaemon_Export(t *testing.T) {
  75. d, start, clean, _, _, _ := mockDaemon(t)
  76. start()
  77. defer clean()
  78. ctx := context.Background()
  79. bytes, err := d.Export(ctx)
  80. if err != nil {
  81. t.Fatalf("Error: %s", err.Error())
  82. }
  83. if string(testBytes) != string(bytes) {
  84. t.Fatalf("Expected %v but got %v", string(testBytes), string(bytes))
  85. }
  86. }
  87. // When I call list workloads, it should list all the workloads
  88. func TestDaemon_ListWorkloads(t *testing.T) {
  89. d, start, clean, _, _, _ := mockDaemon(t)
  90. start()
  91. defer clean()
  92. ctx := context.Background()
  93. // No namespace
  94. s, err := d.ListWorkloads(ctx, "")
  95. if err != nil {
  96. t.Fatalf("Error: %s", err.Error())
  97. }
  98. if len(s) != 2 {
  99. t.Fatalf("Expected %v but got %v", 2, len(s))
  100. }
  101. // Just namespace
  102. s, err = d.ListWorkloads(ctx, ns)
  103. if err != nil {
  104. t.Fatalf("Error: %s", err.Error())
  105. }
  106. if 1 != len(s) {
  107. t.Fatalf("Expected %v but got %v", 1, len(s))
  108. }
  109. // Invalid NS
  110. s, err = d.ListWorkloads(ctx, invalidNS)
  111. if err != nil {
  112. t.Fatalf("Error: %s", err.Error())
  113. }
  114. if len(s) != 0 {
  115. t.Fatalf("Expected %v but got %v", 0, len(s))
  116. }
  117. }
  118. // When I call list workloads with options, it should list all the requested workloads
  119. func TestDaemon_ListWorkloadsWithOptions(t *testing.T) {
  120. d, start, clean, _, _, _ := mockDaemon(t)
  121. start()
  122. defer clean()
  123. ctx := context.Background()
  124. t.Run("no filter", func(t *testing.T) {
  125. s, err := d.ListWorkloadsWithOptions(ctx, v11.ListWorkloadsOptions{})
  126. if err != nil {
  127. t.Fatalf("Error: %s", err.Error())
  128. }
  129. if len(s) != 2 {
  130. t.Fatalf("Expected %v but got %v", 2, len(s))
  131. }
  132. })
  133. t.Run("filter id", func(t *testing.T) {
  134. s, err := d.ListWorkloadsWithOptions(ctx, v11.ListWorkloadsOptions{
  135. Namespace: "",
  136. Services: []flux.ResourceID{flux.MustParseResourceID(wl)}})
  137. if err != nil {
  138. t.Fatalf("Error: %s", err.Error())
  139. }
  140. if len(s) != 1 {
  141. t.Fatalf("Expected %v but got %v", 1, len(s))
  142. }
  143. })
  144. t.Run("filter id and namespace", func(t *testing.T) {
  145. _, err := d.ListWorkloadsWithOptions(ctx, v11.ListWorkloadsOptions{
  146. Namespace: "foo",
  147. Services: []flux.ResourceID{flux.MustParseResourceID(wl)}})
  148. if err == nil {
  149. t.Fatal("Expected error but got nil")
  150. }
  151. })
  152. t.Run("filter unsupported id kind", func(t *testing.T) {
  153. _, err := d.ListWorkloadsWithOptions(ctx, v11.ListWorkloadsOptions{
  154. Namespace: "foo",
  155. Services: []flux.ResourceID{flux.MustParseResourceID("default:unsupportedkind/goodbyeworld")}})
  156. if err == nil {
  157. t.Fatal("Expected error but got nil")
  158. }
  159. })
  160. }
  161. // When I call list images for a workload, it should return images
  162. func TestDaemon_ListImagesWithOptions(t *testing.T) {
  163. d, start, clean, _, _, _ := mockDaemon(t)
  164. start()
  165. defer clean()
  166. ctx := context.Background()
  167. specAll := update.ResourceSpec(update.ResourceSpecAll)
  168. // Service 1
  169. svcID, err := flux.ParseResourceID(wl)
  170. assert.NoError(t, err)
  171. currentImageRef, err := image.ParseRef(currentHelloImage)
  172. assert.NoError(t, err)
  173. newImageRef, err := image.ParseRef(newHelloImage)
  174. assert.NoError(t, err)
  175. oldImageRef, err := image.ParseRef(oldHelloImage)
  176. assert.NoError(t, err)
  177. // Service 2
  178. anotherSvcID, err := flux.ParseResourceID(anotherWl)
  179. assert.NoError(t, err)
  180. anotherImageRef, err := image.ParseRef(anotherImage)
  181. assert.NoError(t, err)
  182. tests := []struct {
  183. name string
  184. opts v10.ListImagesOptions
  185. expectedImages []v6.ImageStatus
  186. expectedNumImages int
  187. shouldError bool
  188. }{
  189. {
  190. name: "All services",
  191. opts: v10.ListImagesOptions{Spec: specAll},
  192. expectedImages: []v6.ImageStatus{
  193. {
  194. ID: svcID,
  195. Containers: []v6.Container{
  196. {
  197. Name: container,
  198. Current: image.Info{ID: currentImageRef},
  199. LatestFiltered: image.Info{ID: newImageRef},
  200. Available: []image.Info{
  201. {ID: newImageRef},
  202. {ID: currentImageRef},
  203. {ID: oldImageRef},
  204. },
  205. AvailableImagesCount: 3,
  206. NewAvailableImagesCount: 1,
  207. FilteredImagesCount: 3,
  208. NewFilteredImagesCount: 1,
  209. },
  210. },
  211. },
  212. {
  213. ID: anotherSvcID,
  214. Containers: []v6.Container{
  215. {
  216. Name: anotherContainer,
  217. Current: image.Info{ID: anotherImageRef},
  218. LatestFiltered: image.Info{},
  219. Available: []image.Info{
  220. {ID: anotherImageRef},
  221. },
  222. AvailableImagesCount: 1,
  223. NewAvailableImagesCount: 0,
  224. FilteredImagesCount: 0, // Excludes latest
  225. NewFilteredImagesCount: 0,
  226. },
  227. },
  228. },
  229. },
  230. shouldError: false,
  231. },
  232. {
  233. name: "Specific service",
  234. opts: v10.ListImagesOptions{Spec: update.ResourceSpec(wl)},
  235. expectedImages: []v6.ImageStatus{
  236. {
  237. ID: svcID,
  238. Containers: []v6.Container{
  239. {
  240. Name: container,
  241. Current: image.Info{ID: currentImageRef},
  242. LatestFiltered: image.Info{ID: newImageRef},
  243. Available: []image.Info{
  244. {ID: newImageRef},
  245. {ID: currentImageRef},
  246. {ID: oldImageRef},
  247. },
  248. AvailableImagesCount: 3,
  249. NewAvailableImagesCount: 1,
  250. FilteredImagesCount: 3,
  251. NewFilteredImagesCount: 1,
  252. },
  253. },
  254. },
  255. },
  256. shouldError: false,
  257. },
  258. {
  259. name: "Override container field selection",
  260. opts: v10.ListImagesOptions{
  261. Spec: specAll,
  262. OverrideContainerFields: []string{"Name", "Current", "NewAvailableImagesCount"},
  263. },
  264. expectedImages: []v6.ImageStatus{
  265. {
  266. ID: svcID,
  267. Containers: []v6.Container{
  268. {
  269. Name: container,
  270. Current: image.Info{ID: currentImageRef},
  271. NewAvailableImagesCount: 1,
  272. },
  273. },
  274. },
  275. {
  276. ID: anotherSvcID,
  277. Containers: []v6.Container{
  278. {
  279. Name: anotherContainer,
  280. Current: image.Info{ID: anotherImageRef},
  281. NewAvailableImagesCount: 0,
  282. },
  283. },
  284. },
  285. },
  286. shouldError: false,
  287. },
  288. {
  289. name: "Override container field selection with invalid field",
  290. opts: v10.ListImagesOptions{
  291. Spec: specAll,
  292. OverrideContainerFields: []string{"InvalidField"},
  293. },
  294. expectedImages: nil,
  295. shouldError: true,
  296. },
  297. {
  298. name: "Specific namespace",
  299. opts: v10.ListImagesOptions{
  300. Spec: specAll,
  301. Namespace: ns,
  302. },
  303. expectedImages: []v6.ImageStatus{
  304. {
  305. ID: svcID,
  306. Containers: []v6.Container{
  307. {
  308. Name: container,
  309. Current: image.Info{ID: currentImageRef},
  310. LatestFiltered: image.Info{ID: newImageRef},
  311. Available: []image.Info{
  312. {ID: newImageRef},
  313. {ID: currentImageRef},
  314. {ID: oldImageRef},
  315. },
  316. AvailableImagesCount: 3,
  317. NewAvailableImagesCount: 1,
  318. FilteredImagesCount: 3,
  319. NewFilteredImagesCount: 1,
  320. },
  321. },
  322. },
  323. },
  324. shouldError: false,
  325. },
  326. {
  327. name: "Specific namespace and service",
  328. opts: v10.ListImagesOptions{
  329. Spec: update.ResourceSpec(wl),
  330. Namespace: ns,
  331. },
  332. expectedImages: nil,
  333. shouldError: true,
  334. },
  335. }
  336. for _, tt := range tests {
  337. t.Run(tt.name, func(t *testing.T) {
  338. is, err := d.ListImagesWithOptions(ctx, tt.opts)
  339. assert.Equal(t, tt.shouldError, err != nil)
  340. // Clear CreatedAt fields for testing
  341. for ri, r := range is {
  342. for ci, c := range r.Containers {
  343. is[ri].Containers[ci].Current.CreatedAt = time.Time{}
  344. is[ri].Containers[ci].LatestFiltered.CreatedAt = time.Time{}
  345. for ai := range c.Available {
  346. is[ri].Containers[ci].Available[ai].CreatedAt = time.Time{}
  347. }
  348. }
  349. }
  350. assert.Equal(t, tt.expectedImages, is)
  351. })
  352. }
  353. }
  354. // When I call notify, it should cause a sync
  355. func TestDaemon_NotifyChange(t *testing.T) {
  356. d, start, clean, mockK8s, events, _ := mockDaemon(t)
  357. w := newWait(t)
  358. ctx := context.Background()
  359. var syncCalled int
  360. var syncDef *cluster.SyncSet
  361. var syncMu sync.Mutex
  362. mockK8s.SyncFunc = func(def cluster.SyncSet) error {
  363. syncMu.Lock()
  364. syncCalled++
  365. syncDef = &def
  366. syncMu.Unlock()
  367. return nil
  368. }
  369. start()
  370. defer clean()
  371. d.NotifyChange(ctx, v9.Change{Kind: v9.GitChange, Source: v9.GitUpdate{}})
  372. w.Eventually(func() bool {
  373. syncMu.Lock()
  374. defer syncMu.Unlock()
  375. return syncCalled == 1
  376. }, "Waiting for sync called")
  377. // Check that sync was called
  378. syncMu.Lock()
  379. defer syncMu.Unlock()
  380. if syncCalled != 1 {
  381. t.Errorf("Sync was not called once, was called %d times", syncCalled)
  382. } else if syncDef == nil {
  383. t.Errorf("Sync was called with a nil syncDef")
  384. }
  385. // Check that history was written to
  386. w.Eventually(func() bool {
  387. es, _ := events.AllEvents(time.Time{}, -1, time.Time{})
  388. for _, e := range es {
  389. if e.Type == event.EventSync {
  390. return true
  391. }
  392. }
  393. return false
  394. }, "Waiting for new sync events")
  395. }
  396. // When I perform a release, it should add a job to update git to the queue
  397. // When I ask about a Job, it should tell me about a job
  398. // When I perform a release, it should update the git repo
  399. func TestDaemon_Release(t *testing.T) {
  400. d, start, clean, _, _, _ := mockDaemon(t)
  401. start()
  402. defer clean()
  403. w := newWait(t)
  404. ctx := context.Background()
  405. // Perform a release
  406. id := updateImage(ctx, d, t)
  407. // Check that job is queued
  408. stat, err := d.JobStatus(ctx, id)
  409. if err != nil {
  410. t.Fatalf("Error: %s", err.Error())
  411. } else if stat.Err != "" {
  412. t.Fatal("Job status error should be empty")
  413. } else if stat.StatusString != job.StatusQueued {
  414. t.Fatalf("Expected %v but got %v", job.StatusQueued, stat.StatusString)
  415. }
  416. // Wait for job to succeed
  417. w.ForJobSucceeded(d, id)
  418. // Wait and check that the git manifest has been altered
  419. w.Eventually(func() bool {
  420. co, err := d.Repo.Clone(ctx, d.GitConfig)
  421. if err != nil {
  422. return false
  423. }
  424. defer co.Clean()
  425. // open a file
  426. dirs := co.ManifestDirs()
  427. if file, err := os.Open(filepath.Join(dirs[0], "helloworld-deploy.yaml")); err == nil {
  428. // make sure it gets closed
  429. defer file.Close()
  430. // create a new scanner and read the file line by line
  431. scanner := bufio.NewScanner(file)
  432. for scanner.Scan() {
  433. if strings.Contains(scanner.Text(), newHelloImage) {
  434. return true
  435. }
  436. }
  437. } else {
  438. t.Fatal(err)
  439. }
  440. // If we get here we haven't found the line we are looking for.
  441. return false
  442. }, "Waiting for new manifest")
  443. }
  444. // When I update a policy, I expect it to add to the queue
  445. // When I update a policy, it should add an annotation to the manifest
  446. func TestDaemon_PolicyUpdate(t *testing.T) {
  447. d, start, clean, _, _, _ := mockDaemon(t)
  448. start()
  449. defer clean()
  450. w := newWait(t)
  451. ctx := context.Background()
  452. // Push an update to a policy
  453. id := updatePolicy(ctx, t, d)
  454. // Wait for job to succeed
  455. w.ForJobSucceeded(d, id)
  456. // Wait and check for new annotation
  457. w.Eventually(func() bool {
  458. co, err := d.Repo.Clone(ctx, d.GitConfig)
  459. if err != nil {
  460. t.Error(err)
  461. return false
  462. }
  463. defer co.Clean()
  464. dirs := co.ManifestDirs()
  465. m, err := d.Manifests.LoadManifests(co.Dir(), dirs)
  466. if err != nil {
  467. t.Fatalf("Error: %s", err.Error())
  468. }
  469. return len(m[wl].Policies()) > 0
  470. }, "Waiting for new annotation")
  471. }
  472. // When I call sync status, it should return a commit showing the sync
  473. // that is about to take place. Then it should return empty once it is
  474. // complete
  475. func TestDaemon_SyncStatus(t *testing.T) {
  476. d, start, clean, _, _, _ := mockDaemon(t)
  477. start()
  478. defer clean()
  479. w := newWait(t)
  480. ctx := context.Background()
  481. // Perform a release
  482. id := updateImage(ctx, d, t)
  483. // Get the commit id
  484. stat := w.ForJobSucceeded(d, id)
  485. // Note: I can't test for an expected number of commits > 0
  486. // because I can't control how fast the sync loop updates the cluster
  487. // Once sync'ed to the cluster, it should empty
  488. w.ForSyncStatus(d, stat.Result.Revision, 0)
  489. }
  490. // When I restart fluxd, there won't be any jobs in the cache
  491. func TestDaemon_JobStatusWithNoCache(t *testing.T) {
  492. d, start, clean, _, _, restart := mockDaemon(t)
  493. start()
  494. defer clean()
  495. w := newWait(t)
  496. ctx := context.Background()
  497. // Perform update
  498. id := updatePolicy(ctx, t, d)
  499. // Make sure the job finishes first
  500. w.ForJobSucceeded(d, id)
  501. // Clear the cache like we've just restarted
  502. restart(func() {
  503. d.JobStatusCache = &job.StatusCache{Size: 100}
  504. })
  505. // Now check if we can get the job status from the commit
  506. w.ForJobSucceeded(d, id)
  507. }
  508. func TestDaemon_Automated(t *testing.T) {
  509. d, start, clean, k8s, _, _ := mockDaemon(t)
  510. start()
  511. defer clean()
  512. w := newWait(t)
  513. workload := cluster.Workload{
  514. ID: flux.MakeResourceID(ns, "deployment", "helloworld"),
  515. Containers: cluster.ContainersOrExcuse{
  516. Containers: []resource.Container{
  517. {
  518. Name: container,
  519. Image: mustParseImageRef(currentHelloImage),
  520. },
  521. },
  522. },
  523. }
  524. k8s.SomeWorkloadsFunc = func([]flux.ResourceID) ([]cluster.Workload, error) {
  525. return []cluster.Workload{workload}, nil
  526. }
  527. // updates from helloworld:master-xxx to helloworld:2
  528. w.ForImageTag(t, d, wl, container, "2")
  529. }
  530. func TestDaemon_Automated_semver(t *testing.T) {
  531. d, start, clean, k8s, _, _ := mockDaemon(t)
  532. start()
  533. defer clean()
  534. w := newWait(t)
  535. resid := flux.MustParseResourceID("default:deployment/semver")
  536. service := cluster.Workload{
  537. ID: resid,
  538. Containers: cluster.ContainersOrExcuse{
  539. Containers: []resource.Container{
  540. {
  541. Name: container,
  542. Image: mustParseImageRef(currentHelloImage),
  543. },
  544. },
  545. },
  546. }
  547. k8s.SomeWorkloadsFunc = func([]flux.ResourceID) ([]cluster.Workload, error) {
  548. return []cluster.Workload{service}, nil
  549. }
  550. // helloworld:3 is older than helloworld:2 but semver orders by version
  551. w.ForImageTag(t, d, resid.String(), container, "3")
  552. }
  553. func makeImageInfo(ref string, t time.Time) image.Info {
  554. return image.Info{ID: mustParseImageRef(ref), CreatedAt: t}
  555. }
  556. func mustParseImageRef(ref string) image.Ref {
  557. r, err := image.ParseRef(ref)
  558. if err != nil {
  559. panic(err)
  560. }
  561. return r
  562. }
  563. type anonNamespacer func(kresource.KubeManifest) string
  564. func (fn anonNamespacer) EffectiveNamespace(m kresource.KubeManifest) (string, error) {
  565. return fn(m), nil
  566. }
  567. var alwaysDefault anonNamespacer = func(kresource.KubeManifest) string {
  568. return "default"
  569. }
  570. func mockDaemon(t *testing.T) (*Daemon, func(), func(), *cluster.Mock, *mockEventWriter, func(func())) {
  571. logger := log.NewNopLogger()
  572. singleService := cluster.Workload{
  573. ID: flux.MustParseResourceID(wl),
  574. Containers: cluster.ContainersOrExcuse{
  575. Containers: []resource.Container{
  576. {
  577. Name: container,
  578. Image: mustParseImageRef(currentHelloImage),
  579. },
  580. },
  581. },
  582. }
  583. multiService := []cluster.Workload{
  584. singleService,
  585. {
  586. ID: flux.MakeResourceID("another", "deployment", "service"),
  587. Containers: cluster.ContainersOrExcuse{
  588. Containers: []resource.Container{
  589. {
  590. Name: anotherContainer,
  591. Image: mustParseImageRef(anotherImage),
  592. },
  593. },
  594. },
  595. },
  596. }
  597. repo, repoCleanup := gittest.Repo(t)
  598. params := git.Config{
  599. Branch: "master",
  600. UserName: "example",
  601. UserEmail: "example@example.com",
  602. SyncTag: "flux-test",
  603. NotesRef: "fluxtest",
  604. }
  605. var k8s *cluster.Mock
  606. {
  607. k8s = &cluster.Mock{}
  608. k8s.AllWorkloadsFunc = func(maybeNamespace string) ([]cluster.Workload, error) {
  609. if maybeNamespace == ns {
  610. return []cluster.Workload{
  611. singleService,
  612. }, nil
  613. } else if maybeNamespace == "" {
  614. return multiService, nil
  615. }
  616. return []cluster.Workload{}, nil
  617. }
  618. k8s.ExportFunc = func() ([]byte, error) { return testBytes, nil }
  619. k8s.PingFunc = func() error { return nil }
  620. k8s.SomeWorkloadsFunc = func([]flux.ResourceID) ([]cluster.Workload, error) {
  621. return []cluster.Workload{
  622. singleService,
  623. }, nil
  624. }
  625. k8s.SyncFunc = func(def cluster.SyncSet) error { return nil }
  626. }
  627. var imageRegistry registry.Registry
  628. {
  629. img0 := makeImageInfo(oldHelloImage, time.Now().Add(-1*time.Second))
  630. img1 := makeImageInfo(currentHelloImage, time.Now())
  631. img2 := makeImageInfo(newHelloImage, time.Now().Add(1*time.Second))
  632. img3 := makeImageInfo("another/service:latest", time.Now().Add(1*time.Second))
  633. imageRegistry = &registryMock.Registry{
  634. Images: []image.Info{
  635. img1,
  636. img2,
  637. img3,
  638. img0,
  639. },
  640. }
  641. }
  642. events := &mockEventWriter{}
  643. // Shutdown chan and waitgroups
  644. jshutdown := make(chan struct{})
  645. dshutdown := make(chan struct{})
  646. jwg := &sync.WaitGroup{}
  647. dwg := &sync.WaitGroup{}
  648. // Jobs queue (starts itself)
  649. jobs := job.NewQueue(jshutdown, jwg)
  650. // Finally, the daemon
  651. d := &Daemon{
  652. Repo: repo,
  653. GitConfig: params,
  654. Cluster: k8s,
  655. Manifests: &kubernetes.Manifests{Namespacer: alwaysDefault},
  656. Registry: imageRegistry,
  657. V: testVersion,
  658. Jobs: jobs,
  659. JobStatusCache: &job.StatusCache{Size: 100},
  660. EventWriter: events,
  661. Logger: logger,
  662. LoopVars: &LoopVars{GitOpTimeout: timeout},
  663. }
  664. start := func() {
  665. if err := repo.Ready(context.Background()); err != nil {
  666. t.Fatal(err)
  667. }
  668. dwg.Add(1)
  669. go d.Loop(dshutdown, dwg, logger)
  670. }
  671. stop := func() {
  672. // Close daemon first so any outstanding jobs are picked up by the queue, otherwise
  673. // calls to Queue#Enqueue() will block forever. Jobs may be enqueued if the daemon's
  674. // image polling picks up automated updates.
  675. close(dshutdown)
  676. dwg.Wait()
  677. close(jshutdown)
  678. jwg.Wait()
  679. repoCleanup()
  680. }
  681. restart := func(f func()) {
  682. close(dshutdown)
  683. dwg.Wait()
  684. f()
  685. dshutdown = make(chan struct{})
  686. start()
  687. }
  688. return d, start, stop, k8s, events, restart
  689. }
  690. type mockEventWriter struct {
  691. events []event.Event
  692. sync.Mutex
  693. }
  694. func (w *mockEventWriter) LogEvent(e event.Event) error {
  695. w.Lock()
  696. defer w.Unlock()
  697. w.events = append(w.events, e)
  698. return nil
  699. }
  700. func (w *mockEventWriter) AllEvents(_ time.Time, _ int64, _ time.Time) ([]event.Event, error) {
  701. w.Lock()
  702. defer w.Unlock()
  703. return w.events, nil
  704. }
  705. // DAEMON TEST HELPERS
  706. type wait struct {
  707. t *testing.T
  708. timeout time.Duration
  709. }
  710. func newWait(t *testing.T) wait {
  711. return wait{
  712. t: t,
  713. timeout: timeout,
  714. }
  715. }
  716. const interval = 10 * time.Millisecond
  717. func (w *wait) Eventually(f func() bool, msg string) {
  718. stop := time.Now().Add(w.timeout)
  719. for time.Now().Before(stop) {
  720. if f() {
  721. return
  722. }
  723. time.Sleep(interval)
  724. }
  725. w.t.Fatal(msg)
  726. }
  727. func (w *wait) ForJobSucceeded(d *Daemon, jobID job.ID) job.Status {
  728. var stat job.Status
  729. var err error
  730. ctx := context.Background()
  731. w.Eventually(func() bool {
  732. stat, err = d.JobStatus(ctx, jobID)
  733. if err != nil {
  734. return false
  735. }
  736. switch stat.StatusString {
  737. case job.StatusSucceeded:
  738. return true
  739. case job.StatusFailed:
  740. w.t.Fatal(stat.Err)
  741. return true
  742. default:
  743. return false
  744. }
  745. }, "Waiting for job to succeed")
  746. return stat
  747. }
  748. func (w *wait) ForSyncStatus(d *Daemon, rev string, expectedNumCommits int) []string {
  749. var revs []string
  750. var err error
  751. w.Eventually(func() bool {
  752. ctx := context.Background()
  753. revs, err = d.SyncStatus(ctx, rev)
  754. return err == nil && len(revs) == expectedNumCommits
  755. }, fmt.Sprintf("Waiting for sync status to have %d commits", expectedNumCommits))
  756. return revs
  757. }
  758. func (w *wait) ForImageTag(t *testing.T, d *Daemon, workload, container, tag string) {
  759. w.Eventually(func() bool {
  760. co, err := d.Repo.Clone(context.TODO(), d.GitConfig)
  761. if err != nil {
  762. return false
  763. }
  764. defer co.Clean()
  765. dirs := co.ManifestDirs()
  766. resources, err := d.Manifests.LoadManifests(co.Dir(), dirs)
  767. assert.NoError(t, err)
  768. workload, ok := resources[workload].(resource.Workload)
  769. assert.True(t, ok)
  770. for _, c := range workload.Containers() {
  771. if c.Name == container && c.Image.Tag == tag {
  772. return true
  773. }
  774. }
  775. return false
  776. }, fmt.Sprintf("Waiting for image tag: %q", tag))
  777. }
  778. func updateImage(ctx context.Context, d *Daemon, t *testing.T) job.ID {
  779. return updateManifest(ctx, t, d, update.Spec{
  780. Type: update.Images,
  781. Spec: update.ReleaseImageSpec{
  782. Kind: update.ReleaseKindExecute,
  783. ServiceSpecs: []update.ResourceSpec{update.ResourceSpecAll},
  784. ImageSpec: newHelloImage,
  785. },
  786. })
  787. }
  788. func updatePolicy(ctx context.Context, t *testing.T, d *Daemon) job.ID {
  789. return updateManifest(ctx, t, d, update.Spec{
  790. Type: update.Policy,
  791. Spec: policy.Updates{
  792. flux.MustParseResourceID("default:deployment/helloworld"): {
  793. Add: policy.Set{
  794. policy.Locked: "true",
  795. },
  796. },
  797. },
  798. })
  799. }
  800. func updateManifest(ctx context.Context, t *testing.T, d *Daemon, spec update.Spec) job.ID {
  801. id, err := d.UpdateManifests(ctx, spec)
  802. if err != nil {
  803. t.Fatalf("Error: %s", err.Error())
  804. }
  805. if id == "" {
  806. t.Fatal("id should not be empty")
  807. }
  808. return id
  809. }