[receiver/github] add concurrency limits and pull request time filtering

Author: adrielp

.chloggen/gh-scrape-limits.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: receiver/github
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add concurrency limit and pull request filtering to reduce rate limiting
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [43388]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

receiver/githubreceiver/README.md

@@ -78,6 +78,8 @@ receivers:
                 github_org: <myfancyorg> 
                 search_query: "org:<myfancyorg> topic:<o11yalltheway>" # Recommended optional query override, defaults to "{org,user}:<github_org>"
                 endpoint: "https://selfmanagedenterpriseserver.com" # Optional
+                concurrency_limit: 50  # Limit to 50 concurrent repos
+                merged_pr_lookback_days: 30  # Only fetch PRs merged in last 30 days
                 auth:
                     authenticator: bearertokenauth/github
 service:
@@ -97,6 +99,10 @@ service:
 
 `search_query` (optional): A filter to narrow down repositories. Defaults to `org:<github_org>` (or `user:<username>`). For example, use `repo:<org>/<repo>` to target a specific repository. Any valid GitHub search syntax is allowed.
 
+`concurrency_limit` (optional): Maximum number of repositories to process concurrently. Defaults to 50. Set to 0 for unlimited (not recommended). GitHub enforces a secondary rate limit of 100 concurrent requests across REST and GraphQL APIs, so values above 100 may cause rate limiting.
+
+`merged_pr_lookback_days` (optional): Limits how far back to look for merged pull requests in days. Defaults to 30 days. Set to 0 to fetch all merged PRs (no filtering). This helps reduce API quota usage by focusing on recent data. Open PRs are always fetched regardless of this setting.
+
 `metrics` (optional): Enable or disable metrics scraping. See the [metrics documentation](./documentation.md) for details.
 
 ### Scraping

receiver/githubreceiver/go.mod

@@ -4,6 +4,7 @@ go 1.24.0
 
 require (
 	github.com/Khan/genqlient v0.8.1
+	github.com/cenkalti/backoff/v5 v5.0.3
 	github.com/google/go-cmp v0.7.0
 	github.com/google/go-github/v79 v79.0.0
 	github.com/gorilla/mux v1.8.1
@@ -35,7 +36,6 @@ require (
 
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/ebitengine/purego v0.9.0 // indirect

receiver/githubreceiver/internal/scraper/githubscraper/README.md

@@ -42,21 +42,16 @@ to prevent abuse and maintain API availability. The following secondary limit is
 particularly relevant:
 
 - **Concurrent Requests Limit**: The API allows no more than 100 concurrent
-requests. This limit is shared across the REST and GraphQL APIs. Since the
-scraper creates a goroutine per repository, having more than 100 repositories
-returned by the `search_query` will result in exceeding this limit.
-It is recommended to use the `search_query` config option to limit the number of
-repositories that are scraped. We recommend one instance of the receiver per
-team (note: `team` is not a valid quantifier when searching repositories `topic`
-is). Reminder that each instance of the receiver should have its own
-corresponding token for authentication as this is what rate limits are tied to.
+  requests. This limit is shared across the REST and GraphQL APIs. The receiver
+  includes a configurable `concurrency_limit` (default 50) to stay under this
+  limit.
 
 In summary, we recommend the following:
 
 - One instance of the receiver per team
 - Each instance of the receiver should have its own token
 - Leverage `search_query` config option to limit repositories returned to 100 or
-less per instance
+less per instance (or configure `concurrency_limit` appropriately)
 - `collection_interval` should be long enough to avoid rate limiting (see above
 formula). A sensible default is `300s`.
 

receiver/githubreceiver/internal/scraper/githubscraper/backoff.go

@@ -0,0 +1,112 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package githubscraper
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/cenkalti/backoff/v5"
+	"go.opentelemetry.io/collector/component"
+	"go.uber.org/zap"
+	semconv "go.opentelemetry.io/otel/semconv/v1.37.0"
+)
+
+// graphqlCallWithRetry wraps a GraphQL API call with exponential backoff
+// retrying on rate limit errors (403, 429) and transient failures
+func graphqlCallWithRetry[T any](
+	ctx context.Context,
+	logger *zap.Logger,
+	telemetry component.TelemetrySettings,
+	scrapeInterval time.Duration,
+	apiCall func() (*T, error),
+) (*T, error) {
+	expBackoff := &backoff.ExponentialBackOff{
+		InitialInterval:     1 * time.Second,
+		RandomizationFactor: backoff.DefaultRandomizationFactor,
+		Multiplier:          backoff.DefaultMultiplier,
+		MaxInterval:         30 * time.Second,
+	}
+
+	var attempts int64
+	operation := func() (*T, error) {
+		attempts++
+		resp, err := apiCall()
+
+		if err == nil {
+			return resp, nil
+		}
+
+		if !isRetriableError(err) {
+			return nil, backoff.Permanent(err)
+		}
+
+		// Log retry attempt with warning level
+		logger.Warn("GitHub API rate limited or temporary failure, retrying",
+			zap.Int64("attempt", attempts),
+			zap.Error(err),
+		)
+
+		// Record retry metric via internal telemetry
+		counter, meterErr := telemetry.MeterProvider.Meter("github.com/open-telemetry/opentelemetry-collector-contrib/receiver/githubreceiver").
+			Int64Counter("http.request.resend_count")
+		if meterErr == nil {
+			counter.Add(ctx, 1)
+		}
+
+		return nil, err
+	}
+
+	// Execute with backoff
+	result, err := backoff.Retry(ctx, operation,
+		backoff.WithBackOff(expBackoff),
+		backoff.WithMaxElapsedTime(scrapeInterval))
+	if err != nil {
+		// Max elapsed time exceeded - log error
+		logger.Error("GitHub API call failed after retries",
+			zap.Int64("total_attempts", attempts),
+			zap.Error(err),
+		)
+
+		// Record failure metric
+		counter, meterErr := telemetry.MeterProvider.Meter("github.com/open-telemetry/opentelemetry-collector-contrib/receiver/githubreceiver").
+			Int64Counter(string(semconv.HTTPRequestResendCountKey))
+		if meterErr == nil {
+			counter.Add(ctx, 1)
+		}
+	}
+
+	return result, err
+}
+
+// isRetriableError determines if an error should trigger a retry
+func isRetriableError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	errStr := err.Error()
+
+	// GraphQL rate limit errors
+	if strings.Contains(errStr, "API rate limit exceeded") {
+		return true
+	}
+
+	// HTTP status code errors
+	if strings.Contains(errStr, "returned error 403") ||
+		strings.Contains(errStr, "returned error 429") ||
+		strings.Contains(errStr, "returned error 502") ||
+		strings.Contains(errStr, "returned error 503") ||
+		strings.Contains(errStr, "returned error 504") {
+		return true
+	}
+
+	// Secondary rate limit
+	if strings.Contains(errStr, "secondary rate limit") {
+		return true
+	}
+
+	return false
+}

receiver/githubreceiver/internal/scraper/githubscraper/config.go

@@ -4,6 +4,8 @@
 package githubscraper // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/githubreceiver/internal/scraper/githubscraper"
 
 import (
+	"errors"
+
 	"go.opentelemetry.io/collector/config/confighttp"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/githubreceiver/internal"
@@ -19,4 +21,18 @@ type Config struct {
 	GitHubOrg string `mapstructure:"github_org"`
 	// SearchQuery is the query to use when defining a custom search for repository data
 	SearchQuery string `mapstructure:"search_query"`
+	// ConcurrencyLimit limits the number of concurrent repository processing goroutines
+	// Default is 50 to stay well under GitHub's 100 concurrent request limit
+	ConcurrencyLimit int `mapstructure:"concurrency_limit"`
+	// MergedPRLookbackDays limits how far back to look for merged pull requests
+	// Default is 30 days. Set to 0 to fetch all merged PRs (backwards compatible)
+	MergedPRLookbackDays int `mapstructure:"merged_pr_lookback_days"`
+}
+
+// Validate validates the configuration
+func (cfg *Config) Validate() error {
+	if cfg.ConcurrencyLimit < 0 {
+		return errors.New("concurrency_limit must be non-negative")
+	}
+	return nil
 }

receiver/githubreceiver/internal/scraper/githubscraper/config_test.go

@@ -25,6 +25,8 @@ func TestConfig(t *testing.T) {
 	expectedConfig := &Config{
 		MetricsBuilderConfig: metadata.DefaultMetricsBuilderConfig(),
 		ClientConfig:         clientConfig,
+		ConcurrencyLimit:     50,
+		MergedPRLookbackDays: 30,
 	}
 
 	assert.Equal(t, expectedConfig, defaultConfig)

receiver/githubreceiver/internal/scraper/githubscraper/factory.go

@@ -30,6 +30,8 @@ func (*Factory) CreateDefaultConfig() internal.Config {
 	return &Config{
 		MetricsBuilderConfig: metadata.DefaultMetricsBuilderConfig(),
 		ClientConfig:         clientConfig,
+		ConcurrencyLimit:     50, // Default to 50 concurrent calls
+		MergedPRLookbackDays: 30, // Default to 30 days
 	}
 }
 
@@ -41,6 +43,11 @@ func (*Factory) CreateMetricsScraper(
 	conf := cfg.(*Config)
 	s := newGitHubScraper(params, conf)
 
+	// // Set scrape interval for backoff max elapsed time
+	// // Default to 30 seconds if not configured
+	// scrapeInterval := 30 * time.Second
+	// s.scrapeInterval = scrapeInterval
+
 	return scraper.NewMetrics(
 		s.scrape,
 		scraper.WithStart(s.start),

receiver/githubreceiver/internal/scraper/githubscraper/github_scraper.go

@@ -24,12 +24,13 @@ import (
 var errClientNotInitErr = errors.New("http client not initialized")
 
 type githubScraper struct {
-	client   *http.Client
-	cfg      *Config
-	settings component.TelemetrySettings
-	logger   *zap.Logger
-	mb       *metadata.MetricsBuilder
-	rb       *metadata.ResourceBuilder
+	client         *http.Client
+	cfg            *Config
+	settings       component.TelemetrySettings
+	logger         *zap.Logger
+	mb             *metadata.MetricsBuilder
+	rb             *metadata.ResourceBuilder
+	scrapeInterval time.Duration
 }
 
 func (ghs *githubScraper) start(ctx context.Context, host component.Host) (err error) {
@@ -95,6 +96,15 @@ func (ghs *githubScraper) scrape(ctx context.Context) (pmetric.Metrics, error) {
 
 	ghs.mb.RecordVcsRepositoryCountDataPoint(now, int64(count))
 
+	// Create semaphore for concurrency limiting
+	var sem chan struct{}
+	if ghs.cfg.ConcurrencyLimit > 0 {
+		sem = make(chan struct{}, ghs.cfg.ConcurrencyLimit)
+		ghs.logger.Debug("Using concurrency limit for repository processing",
+			zap.Int("limit", ghs.cfg.ConcurrencyLimit),
+			zap.Int("total_repos", len(repos)))
+	}
+
 	// Get the ref (branch) count (future branch data) for each repo and record
 	// the given metrics
 	var wg sync.WaitGroup
@@ -107,8 +117,24 @@ func (ghs *githubScraper) scrape(ctx context.Context) (pmetric.Metrics, error) {
 		trunk := repo.DefaultBranchRef.Name
 		now := now
 
+		// Acquire semaphore slot before launching goroutine
+		if sem != nil {
+			select {
+			case sem <- struct{}{}:
+				// Acquired slot, continue
+			case <-ctx.Done():
+				// Context cancelled, skip remaining repos
+				wg.Done()
+				continue
+			}
+		}
+
 		go func() {
 			defer wg.Done()
+			// Release semaphore slot when done
+			if sem != nil {
+				defer func() { <-sem }()
+			}
 
 			branches, count, err := ghs.getBranches(ctx, genClient, name, trunk)
 			if err != nil {

receiver/githubreceiver/internal/scraper/githubscraper/github_scraper_test.go

@@ -91,6 +91,7 @@ func TestScrape(t *testing.T) {
 				},
 				prResponse: prResponse{
 					prs: []getPullRequestDataRepositoryPullRequestsPullRequestConnection{
+						// First call for OPEN PRs
 						{
 							PageInfo: getPullRequestDataRepositoryPullRequestsPullRequestConnectionPageInfo{
 								HasNextPage: false,
@@ -99,6 +100,14 @@ func TestScrape(t *testing.T) {
 								{
 									Merged: false,
 								},
+							},
+						},
+						// Second call for MERGED PRs
+						{
+							PageInfo: getPullRequestDataRepositoryPullRequestsPullRequestConnectionPageInfo{
+								HasNextPage: false,
+							},
+							Nodes: []PullRequestNode{
 								{
 									Merged: true,
 								},
@@ -162,7 +171,10 @@ func TestScrape(t *testing.T) {
 			server := httptest.NewServer(tc.server)
 			defer server.Close()
 
-			cfg := &Config{MetricsBuilderConfig: metadata.DefaultMetricsBuilderConfig()}
+			cfg := &Config{
+				MetricsBuilderConfig: metadata.DefaultMetricsBuilderConfig(),
+				MergedPRLookbackDays: 0, // Fetch all PRs for test
+			}
 
 			ghs := newGitHubScraper(receivertest.NewNopSettings(metadata.Type), cfg)
 			ghs.cfg.GitHubOrg = "open-telemetry"