feat: add concurrency tuning support

add concurrency manager module
update documentation
add args
implement in function logic

Author: BoxBoxJason

README.md

@@ -53,14 +53,17 @@ If mandatory arguments are not provided, the program will prompt for them.
 | Argument | Environment Variable equivalent | Mandatory | Description |
 |----------|-------------------------------|-----------|-------------|
 | `--help` or `-h` | N/A | No | Show help message and exit |
-| `--version` or `-V` | N/A | No | Show version information and exit |
+| `--version` | N/A | No | Show version information and exit |
 | `--verbose` or `-v` | N/A | No | Enable verbose output |
 | `--dry-run` | N/A | No | Perform a dry run without making any changes |
 | `--source-url` | `SOURCE_GITLAB_URL` | Yes | URL of the source GitLab instance |
 | `--source-token` | `SOURCE_GITLAB_TOKEN` | No | Access token for the source GitLab instance |
 | `--destination-url` | `DESTINATION_GITLAB_URL` | Yes | URL of the destination GitLab instance |
 | `--destination-token` | `DESTINATION_GITLAB_TOKEN` | Yes | Access token for the destination GitLab instance |
 | `--mirror-mapping` | `MIRROR_MAPPING` | Yes | Path to a JSON file containing the mirror mapping |
+| `--concurrency` or `-C` | N/A | No | Max number of concurrent requests to the GitLab API (default: 10) |
+| `--timeout` or `-t` | N/A | No | Timeout for GitLab API requests in seconds (default: 30) |
+| `--retry` or `-r` | N/A | No | Number of retries for failed GitLab API requests (does not apply to GraphQL requests) (default: 3) |
 
 ## Example
 

cmd/main.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 
 	"gitlab-sync/mirroring"
 	"gitlab-sync/utils"
@@ -20,15 +21,36 @@ var (
 func main() {
 	var args utils.ParserArgs
 	var mirrorMappingPath string
+	var timeout int
 
 	var rootCmd = &cobra.Command{
-		Use:   "gitlab-sync",
-		Short: "Copy and enable mirroring of gitlab projects and groups",
-		Long:  "Fully customizable gitlab repositories and groups mirroring between two (or one) gitlab instances.",
+		Use:     "gitlab-sync",
+		Version: version,
+		Short:   "Copy and enable mirroring of gitlab projects and groups",
+		Long:    "Fully customizable gitlab repositories and groups mirroring between two (or one) gitlab instances.",
 		Run: func(cmd *cobra.Command, cmdArgs []string) {
-			if args.Version {
-				fmt.Printf("gitlab-sync version: %s\n", version)
-				return
+			// Set the concurrency limit
+			if args.Concurrency == -1 {
+				args.Concurrency = 10000
+			} else if args.Concurrency == 0 {
+				log.Fatal("concurrency limit must be -1 (no limit) or strictly greater than 0")
+			}
+			utils.ConcurrencyManager.SetLimit(args.Concurrency)
+
+			// Obtain the retry count
+			if args.Retry == -1 {
+				args.Retry = 10000
+			} else if args.Retry == 0 {
+				log.Fatal("retry count must be -1 (no limit) or strictly greater than 0")
+			}
+
+			// Set the timeout for GitLab API requests
+			if timeout == -1 {
+				args.Timeout = time.Duration(10000 * time.Second)
+			} else if timeout == 0 {
+				log.Fatal("timeout must be -1 (no limit) or strictly greater than 0")
+			} else {
+				args.Timeout = time.Duration(timeout) * time.Second
 			}
 
 			utils.SetVerbose(args.Verbose)
@@ -65,7 +87,6 @@ func main() {
 		},
 	}
 
-	rootCmd.Flags().BoolVarP(&args.Version, "version", "V", false, "Show version")
 	rootCmd.Flags().StringVar(&args.SourceGitlabURL, "source-url", os.Getenv("SOURCE_GITLAB_URL"), "Source GitLab URL")
 	rootCmd.Flags().StringVar(&args.SourceGitlabToken, "source-token", os.Getenv("SOURCE_GITLAB_TOKEN"), "Source GitLab Token")
 	rootCmd.Flags().StringVar(&args.DestinationGitlabURL, "destination-url", os.Getenv("DESTINATION_GITLAB_URL"), "Destination GitLab URL")
@@ -74,6 +95,9 @@ func main() {
 	rootCmd.Flags().BoolVarP(&args.NoPrompt, "no-prompt", "n", strings.TrimSpace(os.Getenv("NO_PROMPT")) != "", "Disable prompting for missing values")
 	rootCmd.Flags().StringVar(&mirrorMappingPath, "mirror-mapping", os.Getenv("MIRROR_MAPPING"), "Path to the mirror mapping file")
 	rootCmd.Flags().BoolVar(&args.DryRun, "dry-run", false, "Perform a dry run without making any changes")
+	rootCmd.Flags().IntVarP(&timeout, "timeout", "t", 30, "Timeout in seconds for GitLab API requests")
+	rootCmd.Flags().IntVarP(&args.Concurrency, "concurrency", "c", 10, "Max number of concurrent requests")
+	rootCmd.Flags().IntVarP(&args.Retry, "retry", "r", 3, "Number of retries for failed requests")
 
 	if err := rootCmd.Execute(); err != nil {
 		fmt.Println(err)

mirroring/instance.go

@@ -1,7 +1,9 @@
 package mirroring
 
 import (
+	"net/http"
 	"sync"
+	"time"
 
 	"gitlab-sync/utils"
 
@@ -17,8 +19,18 @@ type GitlabInstance struct {
 	GraphQLClient *utils.GraphQLClient
 }
 
-func newGitlabInstance(gitlabURL string, gitlabToken string) (*GitlabInstance, error) {
-	gitlabClient, err := gitlab.NewClient(gitlabToken, gitlab.WithBaseURL(gitlabURL))
+func newGitlabInstance(gitlabURL string, gitlabToken string, timeout time.Duration, maxRetries int) (*GitlabInstance, error) {
+	// Create a custom HTTP client with a timeout
+	httpClient := &http.Client{
+		Timeout: timeout,
+		Transport: &retryTransport{
+			Base:       http.DefaultTransport,
+			MaxRetries: maxRetries,
+		},
+	}
+
+	// Initialize the GitLab client with the custom HTTP client
+	gitlabClient, err := gitlab.NewClient(gitlabToken, gitlab.WithBaseURL(gitlabURL), gitlab.WithHTTPClient(httpClient))
 	if err != nil {
 		return nil, err
 	}
@@ -33,31 +45,56 @@ func newGitlabInstance(gitlabURL string, gitlabToken string) (*GitlabInstance, e
 	return gitlabInstance, nil
 }
 
+// Add a project to the GitLabInstance
 func (g *GitlabInstance) addProject(projectPath string, project *gitlab.Project) {
 	g.muProjects.Lock()
 	defer g.muProjects.Unlock()
 	g.Projects[projectPath] = project
 }
 
+// Get a project from the GitLabInstance
 func (g *GitlabInstance) getProject(projectPath string) *gitlab.Project {
 	g.muProjects.RLock()
 	defer g.muProjects.RUnlock()
-	var project *gitlab.Project
-	project, exists := g.Projects[projectPath]
-	if !exists {
-		project = nil
-	}
-	return project
+	return g.Projects[projectPath]
 }
 
+// Add a group to the GitLabInstance
 func (g *GitlabInstance) addGroup(groupPath string, group *gitlab.Group) {
 	g.muGroups.Lock()
 	defer g.muGroups.Unlock()
 	g.Groups[groupPath] = group
 }
 
+// Get a group from the GitLabInstance
 func (g *GitlabInstance) getGroup(groupPath string) *gitlab.Group {
 	g.muGroups.RLock()
 	defer g.muGroups.RUnlock()
 	return g.Groups[groupPath]
 }
+
+// retryTransport wraps the default HTTP transport to add automatic retries
+type retryTransport struct {
+	Base       http.RoundTripper
+	MaxRetries int
+}
+
+// RoundTrip implements the RoundTripper interface for retryTransport
+func (rt *retryTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	var resp *http.Response
+	var err error
+
+	// Retry the request up to MaxRetries times
+	for i := 0; i <= rt.MaxRetries; i++ {
+		resp, err = rt.Base.RoundTrip(req)
+		if err == nil && resp.StatusCode < http.StatusInternalServerError {
+			// If the request succeeded or returned a non-server-error status, return the response
+			return resp, nil
+		}
+
+		// Retry only on specific server errors or network issues
+		time.Sleep(time.Duration(i) * time.Second) // Exponential backoff
+	}
+
+	return resp, err
+}

mirroring/instance_test.go

@@ -10,7 +10,7 @@ func TestNewGitlabInstance(t *testing.T) {
 	gitlabURL := "https://gitlab.example.com"
 	gitlabToken := "test-token"
 
-	instance, err := newGitlabInstance(gitlabURL, gitlabToken)
+	instance, err := newGitlabInstance(gitlabURL, gitlabToken, 10, 3)
 	if err != nil {
 		t.Fatalf("expected no error, got %v", err)
 	}

mirroring/main.go

@@ -10,12 +10,12 @@ import (
 )
 
 func MirrorGitlabs(gitlabMirrorArgs *utils.ParserArgs) error {
-	sourceGitlabInstance, err := newGitlabInstance(gitlabMirrorArgs.SourceGitlabURL, gitlabMirrorArgs.SourceGitlabToken)
+	sourceGitlabInstance, err := newGitlabInstance(gitlabMirrorArgs.SourceGitlabURL, gitlabMirrorArgs.SourceGitlabToken, gitlabMirrorArgs.Timeout, gitlabMirrorArgs.Retry)
 	if err != nil {
 		return err
 	}
 
-	destinationGitlabInstance, err := newGitlabInstance(gitlabMirrorArgs.DestinationGitlabURL, gitlabMirrorArgs.DestinationGitlabToken)
+	destinationGitlabInstance, err := newGitlabInstance(gitlabMirrorArgs.DestinationGitlabURL, gitlabMirrorArgs.DestinationGitlabToken, gitlabMirrorArgs.Timeout, gitlabMirrorArgs.Retry)
 	if err != nil {
 		return err
 	}

mirroring/post.go

@@ -69,6 +69,7 @@ func createGroups(sourceGitlab *GitlabInstance, destinationGitlab *GitlabInstanc
 
 func createProjects(sourceGitlab *GitlabInstance, destinationGitlab *GitlabInstance, mirrorMapping *utils.MirrorMapping) error {
 	utils.LogVerbose("Creating projects in destination GitLab instance")
+
 	// Reverse the mirror mapping to get the source project path for each destination project
 	reversedMirrorMap := make(map[string]string, len(mirrorMapping.Projects))
 	for sourceProjectPath, projectOptions := range mirrorMapping.Projects {
@@ -81,10 +82,6 @@ func createProjects(sourceGitlab *GitlabInstance, destinationGitlab *GitlabInsta
 	// Create a channel to collect errors
 	errorChan := make(chan error, len(reversedMirrorMap))
 
-	// Create a channel to limit the number of concurrent goroutines
-	concurrencyLimit := 10
-	sem := make(chan struct{}, concurrencyLimit)
-
 	for destinationProjectPath, sourceProjectPath := range reversedMirrorMap {
 		utils.LogVerbosef("Mirroring project from source %s to destination %s", sourceProjectPath, destinationProjectPath)
 		sourceProject := sourceGitlab.Projects[sourceProjectPath]
@@ -93,13 +90,9 @@ func createProjects(sourceGitlab *GitlabInstance, destinationGitlab *GitlabInsta
 			continue
 		}
 		wg.Add(1)
-		// Acquire a token from the semaphore
-		sem <- struct{}{}
 
 		go func(sourcePath string, destinationPath string) {
 			defer wg.Done()
-			// Release the token back to the semaphore
-			defer func() { <-sem }()
 
 			// Retrieve the corresponding project creation options from the mirror mapping
 			projectCreationOptions, ok := mirrorMapping.Projects[sourcePath]
@@ -136,42 +129,64 @@ func createProjects(sourceGitlab *GitlabInstance, destinationGitlab *GitlabInsta
 		}(sourceProjectPath, destinationProjectPath)
 	}
 
-	// Wait for all goroutines to finish
 	wg.Wait()
 	close(errorChan)
 
 	return utils.MergeErrors(errorChan, 2)
 }
 
 func (g *GitlabInstance) createProjectFromSource(sourceProject *gitlab.Project, copyOptions *utils.MirroringOptions) (*gitlab.Project, error) {
-	projectCreationArgs := &gitlab.CreateProjectOptions{
-		Name:                &sourceProject.Name,
-		Path:                &sourceProject.Path,
-		DefaultBranch:       &sourceProject.DefaultBranch,
-		Description:         &sourceProject.Description,
-		MirrorTriggerBuilds: gitlab.Ptr(copyOptions.MirrorTriggerBuilds),
-		Mirror:              gitlab.Ptr(true),
-		Topics:              &sourceProject.Topics,
-		Visibility:          gitlab.Ptr(gitlab.VisibilityValue(copyOptions.Visibility)),
-	}
+	// Create a wait group and error channel for error handling
+	var wg sync.WaitGroup
+	errorChan := make(chan error, 1)
+
+	// Define the API call logic
+	apiFunc := func() error {
+		projectCreationArgs := &gitlab.CreateProjectOptions{
+			Name:                &sourceProject.Name,
+			Path:                &sourceProject.Path,
+			DefaultBranch:       &sourceProject.DefaultBranch,
+			Description:         &sourceProject.Description,
+			MirrorTriggerBuilds: &copyOptions.MirrorTriggerBuilds,
+			Mirror:              gitlab.Ptr(true),
+			Topics:              &sourceProject.Topics,
+			Visibility:          gitlab.Ptr(gitlab.VisibilityValue(copyOptions.Visibility)),
+		}
 
-	utils.LogVerbosef("Retrieving project namespace ID for %s", copyOptions.DestinationPath)
-	parentNamespaceId, err := g.getParentNamespaceID(copyOptions.DestinationPath)
-	if err != nil {
-		return nil, err
-	} else if parentNamespaceId >= 0 {
-		projectCreationArgs.NamespaceID = &parentNamespaceId
+		utils.LogVerbosef("Retrieving project namespace ID for %s", copyOptions.DestinationPath)
+		parentNamespaceId, err := g.getParentNamespaceID(copyOptions.DestinationPath)
+		if err != nil {
+			return err
+		} else if parentNamespaceId >= 0 {
+			projectCreationArgs.NamespaceID = &parentNamespaceId
+		}
+
+		utils.LogVerbosef("Creating project %s in destination GitLab instance", copyOptions.DestinationPath)
+		destinationProject, _, err := g.Gitlab.Projects.CreateProject(projectCreationArgs)
+		if err != nil {
+			return err
+		}
+		utils.LogVerbosef("Project %s created successfully", destinationProject.PathWithNamespace)
+		g.addProject(copyOptions.DestinationPath, destinationProject)
+
+		return nil
 	}
 
-	utils.LogVerbosef("Creating project %s in destination GitLab instance", copyOptions.DestinationPath)
-	destinationProject, _, err := g.Gitlab.Projects.CreateProject(projectCreationArgs)
-	if err != nil {
+	// Increment the wait group counter and execute the API call
+	wg.Add(1)
+	go utils.ExecuteWithConcurrency(apiFunc, &wg, errorChan)
+
+	// Wait for the API call to complete
+	wg.Wait()
+	close(errorChan)
+
+	// Check for errors
+	select {
+	case err := <-errorChan:
 		return nil, err
+	default:
+		return nil, nil
 	}
-	utils.LogVerbosef("Project %s created successfully", destinationProject.PathWithNamespace)
-	g.addProject(copyOptions.DestinationPath, destinationProject)
-
-	return destinationProject, nil
 }
 
 func (g *GitlabInstance) createGroupFromSource(sourceGroup *gitlab.Group, copyOptions *utils.MirroringOptions) (*gitlab.Group, error) {
@@ -219,6 +234,10 @@ func (g *GitlabInstance) mirrorReleases(sourceProject *gitlab.Project, destinati
 		return fmt.Errorf("failed to fetch releases for source project %s: %s", sourceProject.PathWithNamespace, err)
 	}
 
+	// Create a wait group and an error channel for handling API calls concurrently
+	var wg sync.WaitGroup
+	errorChan := make(chan error, len(sourceReleases))
+
 	// Iterate over each source release
 	for _, release := range sourceReleases {
 		// Check if the release already exists in the destination project
@@ -227,20 +246,48 @@ func (g *GitlabInstance) mirrorReleases(sourceProject *gitlab.Project, destinati
 			continue
 		}
 
-		utils.LogVerbosef("Mirroring release %s to project %s", release.TagName, destinationProject.PathWithNamespace)
+		// Increment the wait group counter
+		wg.Add(1)
 
-		// Create the release in the destination project
-		_, _, err := g.Gitlab.Releases.CreateRelease(destinationProject.ID, &gitlab.CreateReleaseOptions{
-			Name:        gitlab.Ptr(release.Name),
-			TagName:     gitlab.Ptr(release.TagName),
-			Description: gitlab.Ptr(release.Description),
-			ReleasedAt:  release.ReleasedAt,
-		})
-		if err != nil {
-			utils.LogVerbosef("Failed to create release %s in project %s: %s", release.TagName, destinationProject.PathWithNamespace, err)
+		// Define the API call logic for creating a release
+		releaseToMirror := release // Capture the current release in the loop
+		go utils.ExecuteWithConcurrency(func() error {
+			utils.LogVerbosef("Mirroring release %s to project %s", releaseToMirror.TagName, destinationProject.PathWithNamespace)
+
+			// Create the release in the destination project
+			_, _, err := g.Gitlab.Releases.CreateRelease(destinationProject.ID, &gitlab.CreateReleaseOptions{
+				Name:        &releaseToMirror.Name,
+				TagName:     &releaseToMirror.TagName,
+				Description: &releaseToMirror.Description,
+				ReleasedAt:  releaseToMirror.ReleasedAt,
+			})
+			if err != nil {
+				utils.LogVerbosef("Failed to create release %s in project %s: %s", releaseToMirror.TagName, destinationProject.PathWithNamespace, err)
+				return fmt.Errorf("Failed to create release %s in project %s: %s", releaseToMirror.TagName, destinationProject.PathWithNamespace, err)
+			}
+
+			return nil
+		}, &wg, errorChan)
+	}
+
+	// Wait for all goroutines to finish
+	wg.Wait()
+	close(errorChan)
+
+	// Check the error channel for any errors
+	var combinedError error
+	for err := range errorChan {
+		if combinedError == nil {
+			combinedError = err
+		} else {
+			combinedError = fmt.Errorf("%s; %s", combinedError, err)
 		}
 	}
 
+	if combinedError != nil {
+		return combinedError
+	}
+
 	utils.LogVerbosef("Releases mirroring completed for project %s", destinationProject.HTTPURLToRepo)
 	return nil
 }