Add UT cases for package opm (update travis yaml to start redis container)

Fix data race issue in opm package

.travis.yml

@@ -28,11 +28,13 @@ env:
   AUTH_MODE: db_auth
   SELF_REGISTRATION: on
   KEY_PATH: /data/secretkey
+  REDIS_HOST: localhost
 
 before_install:
    - sudo ./tests/hostcfg.sh
    - sudo ./tests/generateCerts.sh
    - sudo ./make/prepare
+   - sudo mkdir -p "/data/redis"
 
 install: 
   - sudo apt-get update && sudo apt-get install -y libldap2-dev 
@@ -114,6 +116,7 @@ script:
   - ./tests/swaggerchecker.sh
   - ./tests/startuptest.sh
   - ./tests/userlogintest.sh ${HARBOR_ADMIN} ${HARBOR_ADMIN_PASSWD}
+  - export REDIS_HOST=$IP
 
 #  - sudo ./tests/testprepare.sh
 #  - go test -v ./tests/apitests

src/jobservice_v2/opm/redis_job_stats_mgr.go

@@ -10,6 +10,7 @@ import (
 	"math"
 	"math/rand"
 	"strconv"
+	"sync/atomic"
 	"time"
 
 	"github.com/vmware/harbor/src/jobservice_v2/errs"
@@ -55,12 +56,15 @@ type RedisJobStatsManager struct {
 	stopChan    chan struct{}
 	doneChan    chan struct{}
 	processChan chan *queueItem
-	isRunning   bool       //no need to sync
+	isRunning   *atomic.Value
 	hookStore   *HookStore //cache the hook here to avoid requesting backend
 }
 
 //NewRedisJobStatsManager is constructor of RedisJobStatsManager
 func NewRedisJobStatsManager(ctx context.Context, namespace string, redisPool *redis.Pool) *RedisJobStatsManager {
+	isRunning := &atomic.Value{}
+	isRunning.Store(false)
+
 	return &RedisJobStatsManager{
 		namespace:   namespace,
 		context:     ctx,
@@ -69,16 +73,17 @@ func NewRedisJobStatsManager(ctx context.Context, namespace string, redisPool *r
 		doneChan:    make(chan struct{}, 1),
 		processChan: make(chan *queueItem, processBufferSize),
 		hookStore:   NewHookStore(),
+		isRunning:   isRunning,
 	}
 }
 
 //Start is implementation of same method in JobStatsManager interface.
 func (rjs *RedisJobStatsManager) Start() {
-	if rjs.isRunning {
+	if rjs.isRunning.Load().(bool) {
 		return
 	}
 	go rjs.loop()
-	rjs.isRunning = true
+	rjs.isRunning.Store(true)
 
 	logger.Info("Redis job stats manager is started")
 }
@@ -86,10 +91,10 @@ func (rjs *RedisJobStatsManager) Start() {
 //Shutdown is implementation of same method in JobStatsManager interface.
 func (rjs *RedisJobStatsManager) Shutdown() {
 	defer func() {
-		rjs.isRunning = false
+		rjs.isRunning.Store(false)
 	}()
 
-	if !rjs.isRunning {
+	if !(rjs.isRunning.Load().(bool)) {
 		return
 	}
 	rjs.stopChan <- struct{}{}
@@ -139,7 +144,7 @@ func (rjs *RedisJobStatsManager) loop() {
 	controlChan := make(chan struct{})
 
 	defer func() {
-		rjs.isRunning = false
+		rjs.isRunning.Store(false)
 		//Notify other sub goroutines
 		close(controlChan)
 		logger.Info("Redis job stats manager is stopped")
@@ -153,6 +158,8 @@ func (rjs *RedisJobStatsManager) loop() {
 				if err := rjs.process(item); err != nil {
 					item.fails++
 					if item.fails < maxFails {
+						logger.Warningf("Failed to process '%s' request with error: %s\n", item.op, err)
+
 						//Retry after a random interval
 						go func() {
 							timer := time.NewTimer(time.Duration(backoff(item.fails)) * time.Second)

src/jobservice_v2/opm/redis_job_stats_mgr_test.go

@@ -1,6 +1,269 @@
 // Copyright 2018 The Harbor Authors. All rights reserved.
 package opm
 
-import "testing"
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+	"time"
 
-func TestRetrieveJob(t *testing.T) {}
+	"github.com/garyburd/redigo/redis"
+	"github.com/vmware/harbor/src/jobservice_v2/job"
+	"github.com/vmware/harbor/src/jobservice_v2/models"
+	"github.com/vmware/harbor/src/jobservice_v2/utils"
+)
+
+const (
+	dialConnectionTimeout = 30 * time.Second
+	healthCheckPeriod     = time.Minute
+	dialReadTimeout       = healthCheckPeriod + 10*time.Second
+	dialWriteTimeout      = 10 * time.Second
+	testingRedisHost      = "REDIS_HOST"
+	testingNamespace      = "testing_job_service_v2"
+)
+
+var redisHost = getRedisHost()
+var redisPool = &redis.Pool{
+	MaxActive: 2,
+	MaxIdle:   2,
+	Wait:      true,
+	Dial: func() (redis.Conn, error) {
+		return redis.Dial(
+			"tcp",
+			fmt.Sprintf("%s:%d", redisHost, 6379),
+			redis.DialConnectTimeout(dialConnectionTimeout),
+			redis.DialReadTimeout(dialReadTimeout),
+			redis.DialWriteTimeout(dialWriteTimeout),
+		)
+	},
+}
+
+func TestSetJobStatus(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+	//make sure data existing
+	testingStats := createFakeStats()
+	mgr.Save(testingStats)
+	<-time.After(200 * time.Millisecond)
+
+	mgr.SetJobStatus("fake_job_ID", "running")
+	<-time.After(100 * time.Millisecond)
+	stats, err := mgr.Retrieve("fake_job_ID")
+	if err != nil {
+		t.Error(err)
+	}
+
+	if stats.Stats.Status != "running" {
+		t.Errorf("expect job status 'running' but got '%s'\n", stats.Stats.Status)
+	}
+
+	key := utils.KeyJobStats(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestCommand(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+
+	if err := mgr.SendCommand("fake_job_ID", CtlCommandStop); err != nil {
+		t.Error(err)
+	}
+
+	if cmd, err := mgr.CtlCommand("fake_job_ID"); err != nil {
+		t.Error(err)
+	} else {
+		if cmd != CtlCommandStop {
+			t.Errorf("expect '%s' but got '%s'", CtlCommandStop, cmd)
+		}
+	}
+
+	key := utils.KeyJobCtlCommands(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestDieAt(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+
+	testingStats := createFakeStats()
+	mgr.Save(testingStats)
+
+	dieAt := time.Now().Unix()
+	if err := createDeadJob(redisPool.Get(), dieAt); err != nil {
+		t.Error(err)
+	}
+	<-time.After(200 * time.Millisecond)
+	mgr.DieAt("fake_job_ID", dieAt)
+	<-time.After(300 * time.Millisecond)
+
+	stats, err := mgr.Retrieve("fake_job_ID")
+	if err != nil {
+		t.Error(err)
+	}
+
+	if stats.Stats.DieAt != dieAt {
+		t.Errorf("expect die at '%d' but got '%d'\n", dieAt, stats.Stats.DieAt)
+	}
+
+	key := utils.KeyJobStats(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+	key2 := utils.RedisKeyDead(testingNamespace)
+	if err := clear(key2, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestRegisterHook(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+
+	if err := mgr.RegisterHook("fake_job_ID", "http://localhost:9999", false); err != nil {
+		t.Error(err)
+	}
+
+	key := utils.KeyJobStats(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestExpireJobStats(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+
+	//make sure data existing
+	testingStats := createFakeStats()
+	mgr.Save(testingStats)
+	<-time.After(200 * time.Millisecond)
+
+	if err := mgr.ExpirePeriodicJobStats("fake_job_ID"); err != nil {
+		t.Error(err)
+	}
+
+	key := utils.KeyJobStats(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestCheckIn(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+
+	//make sure data existing
+	testingStats := createFakeStats()
+	mgr.Save(testingStats)
+	<-time.After(200 * time.Millisecond)
+
+	//Start http server
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintln(w, "ok")
+	}))
+	defer ts.Close()
+
+	if err := mgr.RegisterHook("fake_job_ID", ts.URL, false); err != nil {
+		t.Error(err)
+	}
+
+	mgr.CheckIn("fake_job_ID", "checkin")
+	<-time.After(200 * time.Millisecond)
+
+	stats, err := mgr.Retrieve("fake_job_ID")
+	if err != nil {
+		t.Error(err)
+	}
+
+	if stats.Stats.CheckIn != "checkin" {
+		t.Errorf("expect check in info 'checkin' but got '%s'\n", stats.Stats.CheckIn)
+	}
+
+	key := utils.KeyJobStats(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func getRedisHost() string {
+	redisHost := os.Getenv(testingRedisHost)
+	if redisHost == "" {
+		redisHost = "10.160.178.186" //for local test
+	}
+
+	return redisHost
+}
+
+func createStatsManager(redisPool *redis.Pool) JobStatsManager {
+	ctx := context.Background()
+	return NewRedisJobStatsManager(ctx, testingNamespace, redisPool)
+}
+
+func clear(key string, conn redis.Conn) error {
+	if conn != nil {
+		defer conn.Close()
+		_, err := conn.Do("DEL", key)
+		return err
+	}
+
+	return errors.New("failed to clear")
+}
+
+func createFakeStats() models.JobStats {
+	testingStats := models.JobStats{
+		Stats: &models.JobStatData{
+			JobID:       "fake_job_ID",
+			JobKind:     job.JobKindPeriodic,
+			JobName:     "fake_job",
+			Status:      "Pending",
+			IsUnique:    false,
+			RefLink:     "/api/v1/jobs/fake_job_ID",
+			CronSpec:    "5 * * * * *",
+			EnqueueTime: time.Now().Unix(),
+			UpdateTime:  time.Now().Unix(),
+		},
+	}
+
+	return testingStats
+}
+
+func createDeadJob(conn redis.Conn, dieAt int64) error {
+	dead := make(map[string]interface{})
+	dead["name"] = "fake_job"
+	dead["id"] = "fake_job_ID"
+	dead["args"] = make(map[string]interface{})
+	dead["fails"] = 3
+	dead["err"] = "testing error"
+	dead["failed_at"] = dieAt
+
+	rawJSON, err := json.Marshal(&dead)
+	if err != nil {
+		return err
+	}
+
+	defer conn.Close()
+	key := utils.RedisKeyDead(testingNamespace)
+	_, err = conn.Do("ZADD", key, dieAt, rawJSON)
+	return err
+}

tests/docker-compose.test.yml

@@ -34,3 +34,10 @@ services:
       - /data/:/data/
     ports:
       - 8888:8080
+  redis:
+    image: vmware/redis-photon:4.0
+    restart: always
+    volumes:
+      - /data/redis:/data
+    ports:
+      - 6379:6379