Merge branch 'main' into dependabot/github_actions/google-github-acti…

…ons/setup-gcloud-1

Makefile

@@ -107,8 +107,8 @@ PREPARE_VERSION_NAME=versions
 REGISTRYVERSION=v2.8.0-patch-redis
 NOTARYVERSION=v0.6.1
 NOTARYMIGRATEVERSION=v4.11.0
-TRIVYVERSION=v0.37.2
-TRIVYADAPTERVERSION=v0.30.7
+TRIVYVERSION=v0.38.2
+TRIVYADAPTERVERSION=v0.30.9
 
 # version of registry for pulling the source code
 REGISTRY_SRC_TAG=v2.8.0

make/migrations/postgresql/0110_2.8.0_schema.up.sql

@@ -91,41 +91,29 @@ SET enabled = false,
     description = 'Chartmuseum is deprecated in Harbor v2.8.0, because this notification policy only has event type about Chartmuseum, so please update or delete this notification policy.'
 WHERE event_types = '[]';
 
-/* insert the default payload_format for http type webhook target */
+/* insert the default payload_format for http type webhook target
+1. separate the original targets(text) to json array elements(targets_expanded)
+2. update the old target to set the payload format if type is 'http' into the targets_updated
+3. finally update back to the original table notification_policy
+*/
 WITH targets_expanded AS (
-    -- Expand the JSON array of targets into separate rows
-    SELECT
-        id,
-        jsonb_array_elements(targets::jsonb) AS target
-    FROM
-        notification_policy
+    SELECT id, jsonb_array_elements(targets::jsonb) AS target
+    FROM notification_policy
 ),
 targets_updated AS (
-    -- Update targets based on the specified conditions
-    SELECT
-        id,
+    SELECT id,
         jsonb_agg(
             CASE
-                -- If target is HTTP and has no payload format, add "Default"
                 WHEN target->>'type' = 'http' AND NOT target ? 'payload_format'
                 THEN target || '{"payload_format":"Default"}'::jsonb
                 ELSE target
             END
         ) AS targets
-    FROM
-        targets_expanded
-    GROUP BY
-        id
+    FROM targets_expanded GROUP BY id
 )
--- Update the original table with the updated targets
-UPDATE
-    notification_policy
-SET
-    targets = targets_updated.targets
-FROM
-    targets_updated
-WHERE
-    notification_policy.id = targets_updated.id;
+UPDATE notification_policy
+SET targets = targets_updated.targets
+FROM targets_updated WHERE notification_policy.id = targets_updated.id;
 
 /* migrate the webhook job to execution and task as the webhook refactor since v2.8 */
 DO $$

src/common/const.go

@@ -224,4 +224,6 @@ const (
 
 	// UIMaxLengthLimitedOfNumber is the max length that UI limited for type number
 	UIMaxLengthLimitedOfNumber = 10
+	// ExecutionStatusRefreshIntervalSeconds is the interval seconds for refreshing execution status
+	ExecutionStatusRefreshIntervalSeconds = "execution_status_refresh_interval_seconds"
 )

src/core/main.go

@@ -49,6 +49,7 @@ import (
 	_ "github.com/goharbor/harbor/src/lib/cache/memory" // memory cache
 	_ "github.com/goharbor/harbor/src/lib/cache/redis"  // redis cache
 	"github.com/goharbor/harbor/src/lib/config"
+	"github.com/goharbor/harbor/src/lib/gtask"
 	"github.com/goharbor/harbor/src/lib/log"
 	"github.com/goharbor/harbor/src/lib/metric"
 	"github.com/goharbor/harbor/src/lib/orm"
@@ -207,6 +208,9 @@ func main() {
 	health.RegisterHealthCheckers()
 	registerScanners(orm.Context())
 
+	// start global task pool, do not stop in the gracefulShutdown because it may take long time to finish.
+	gtask.DefaultPool().Start(ctx)
+
 	closing := make(chan struct{})
 	done := make(chan struct{})
 	go gracefulShutdown(closing, done, shutdownTracerProvider)

src/lib/config/metadata/metadatalist.go

@@ -189,5 +189,7 @@ var (
 		{Name: common.ScannerSkipUpdatePullTime, Scope: UserScope, Group: BasicGroup, EnvKey: "SCANNER_SKIP_UPDATE_PULL_TIME", DefaultValue: "false", ItemType: &BoolType{}, Editable: false, Description: `The option to skip update pull time for scanner`},
 
 		{Name: common.SessionTimeout, Scope: UserScope, Group: BasicGroup, EnvKey: "SESSION_TIMEOUT", DefaultValue: "60", ItemType: &Int64Type{}, Editable: true, Description: `The session timeout in minutes`},
+
+		{Name: common.ExecutionStatusRefreshIntervalSeconds, Scope: SystemScope, Group: BasicGroup, EnvKey: "EXECUTION_STATUS_REFRESH_INTERVAL_SECONDS", DefaultValue: "30", ItemType: &Int64Type{}, Editable: false, Description: `The interval seconds to refresh the execution status`},
 	}
 )

src/lib/config/systemconfig.go

@@ -141,6 +141,11 @@ func GetGCTimeWindow() int64 {
 	return common.DefaultGCTimeWindowHours
 }
 
+// GetExecutionStatusRefreshIntervalSeconds returns the interval seconds for the refresh of execution status.
+func GetExecutionStatusRefreshIntervalSeconds() int64 {
+	return DefaultMgr().Get(backgroundCtx, common.ExecutionStatusRefreshIntervalSeconds).GetInt64()
+}
+
 // WithNotary returns a bool value to indicate if Harbor's deployed with Notary
 func WithNotary() bool {
 	return DefaultMgr().Get(backgroundCtx, common.WithNotary).GetBool()

src/lib/gtask/pool.go

@@ -0,0 +1,97 @@
+// Copyright Project Harbor Authors
+//
+// 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 gtask
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+func DefaultPool() *Pool {
+	return pool
+}
+
+var (
+	// pool is the global task pool.
+	pool = NewPool()
+)
+
+type taskFunc func(ctx context.Context)
+
+// Pool is the task pool for managing some async jobs.
+type Pool struct {
+	stopCh chan struct{}
+	wg     sync.WaitGroup
+	lock   sync.Mutex
+	tasks  []*task
+}
+
+func NewPool() *Pool {
+	return &Pool{
+		stopCh: make(chan struct{}),
+	}
+}
+
+type task struct {
+	fn       taskFunc
+	interval time.Duration
+}
+
+func (p *Pool) AddTask(fn taskFunc, interval time.Duration) {
+	t := &task{
+		fn:       fn,
+		interval: interval,
+	}
+
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	p.tasks = append(p.tasks, t)
+}
+
+func (p *Pool) Start(ctx context.Context) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	for _, task := range p.tasks {
+		p.wg.Add(1)
+		go p.doTask(ctx, task)
+	}
+}
+
+func (p *Pool) doTask(ctx context.Context, task *task) {
+	defer p.wg.Done()
+	for {
+		select {
+		// wait for stop signal
+		case <-ctx.Done():
+			return
+		case <-p.stopCh:
+			return
+		default:
+			task.fn(ctx)
+			// interval is 0 means it's a one time job, return directly
+			if task.interval == 0 {
+				return
+			}
+			time.Sleep(task.interval)
+		}
+	}
+}
+
+func (p *Pool) Stop() {
+	close(p.stopCh)
+	p.wg.Wait()
+}

src/lib/gtask/pool_test.go

@@ -0,0 +1,103 @@
+// Copyright Project Harbor Authors
+//
+// 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 gtask
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAddTask(t *testing.T) {
+	pool := NewPool()
+
+	taskNum := 3
+	taskInterval := time.Duration(0)
+	for i := 0; i < taskNum; i++ {
+		fn := func(ctx context.Context) {
+			t.Logf("Task %d is running...", i)
+		}
+
+		pool.AddTask(fn, taskInterval)
+	}
+
+	if len(pool.tasks) != taskNum {
+		t.Errorf("Expected %d tasks but found %d", taskNum, len(pool.tasks))
+	}
+}
+
+func TestStartAndStop(t *testing.T) {
+	// test normal case
+	{
+		pool := NewPool()
+		// create channel with buffer
+		ch1 := make(chan struct{}, 5)
+		ch2 := make(chan struct{}, 5)
+		// test one-time job
+		t1 := &task{
+			interval: 0,
+			fn: func(ctx context.Context) {
+				ch1 <- struct{}{}
+			},
+		}
+		// test interval job
+		t2 := &task{
+			interval: 100 * time.Millisecond,
+			fn: func(ctx context.Context) {
+				ch2 <- struct{}{}
+			},
+		}
+
+		pool.tasks = []*task{t1, t2}
+
+		ctx1, cancel1 := context.WithCancel(context.Background())
+		defer cancel1()
+		pool.Start(ctx1)
+
+		// Let it run for a bit
+		time.Sleep(300 * time.Millisecond)
+		// ch1 should only have one element as it's a one time job
+		assert.Equal(t, 1, len(ch1))
+		// ch2 should have elements over 2 as sleep 300ms and interval is 100ms
+		assert.Greater(t, len(ch2), 2)
+		pool.Stop()
+		close(ch1)
+		close(ch2)
+	}
+
+	// test context timeout case
+	{
+		pool := NewPool()
+		ch1 := make(chan struct{}, 2)
+		t1 := &task{
+			interval: 100 * time.Millisecond,
+			fn: func(ctx context.Context) {
+				ch1 <- struct{}{}
+			},
+		}
+
+		pool.tasks = []*task{t1}
+		ctx1, cancel1 := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel1()
+		pool.Start(ctx1)
+		// Let it run for a bit
+		time.Sleep(200 * time.Millisecond)
+		assert.Equal(t, 1, len(ch1))
+		pool.Stop()
+		close(ch1)
+	}
+}

src/lib/shuffle.go

@@ -0,0 +1,28 @@
+// Copyright Project Harbor Authors
+//
+// 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 lib
+
+import (
+	"math/rand"
+	"time"
+)
+
+// ShuffleStringSlice shuffles the string slice in place.
+func ShuffleStringSlice(slice []string) {
+	rd := rand.New(rand.NewSource(time.Now().UnixNano()))
+	rd.Shuffle(len(slice), func(i, j int) {
+		slice[i], slice[j] = slice[j], slice[i]
+	})
+}