README
¶
Crawler Documentation
This directory contains the core crawling functionality and resilience features.
Core Components
Crawler (crawler.go)
The main crawler service that orchestrates web scraping operations.
Features:
- Worker pool for concurrent processing
- Automatic retry with exponential backoff
- Content extraction (title, text, links, images)
- Integration with all queue backends
- Comprehensive metrics collection
Basic Usage:
crawler := crawler.NewCrawler(queue, storage, nil, nil, 10, 30*time.Second)
crawler.Start()
defer crawler.Stop()
Kafka Enhanced Crawler (kafka_enhanced_crawler.go)
Extended crawler with Kafka-specific features for high-scale deployments.
Additional Features:
- Event sourcing for complete audit trail
- Priority-based processing
- Dead letter queue handling
- Batch URL processing
- Advanced metrics publishing
Usage:
enhancedCrawler := crawler.NewKafkaEnhancedCrawler(baseCrawler, producerService)
enhancedCrawler.StartEnhanced()
Resilience Features
Circuit Breaker (circuitbreaker.go)
Protects against failing domains to prevent resource waste.
States:
- Closed: Normal operation, requests pass through
- Open: Failing domain blocked, requests fail fast
- Half-Open: Testing if domain has recovered
Configuration:
breaker := circuitbreaker.New(circuitbreaker.Config{
FailureThreshold: 5, // Failures before opening
RecoveryTimeout: 60*time.Second, // Time before retry
SuccessThreshold: 3, // Successes to close
})
Usage:
err := breaker.Execute(func() error {
return scrapeURL(url)
})
Rate Limiter (ratelimiter.go)
Per-host rate limiting to respect target websites and avoid IP bans.
Features:
- Individual rate limits per host
- Configurable requests per second and burst capacity
- Automatic cleanup of unused limiters
- Thread-safe implementation
Configuration:
limiter := ratelimiter.NewHostRateLimiter(0.2, 1) // 1 request per 5 seconds, burst of 1
Usage:
if limiter.Allow(host) {
// Proceed with request
response, err := scrapeURL(url)
}
Robots.txt Handler (robots.go)
Respects website robots.txt policies for ethical scraping.
Features:
- Automatic robots.txt fetching and caching
- Per-host robots.txt compliance
- Configurable user agent
- TTL-based cache expiration
Usage:
robotsCache := robots.NewRobotsCache("WebScraper/1.0")
if robotsCache.IsAllowed(url) {
// Safe to scrape this URL
}
Content Extraction
The crawler automatically extracts:
Title Extraction
title := extractTitleFromHTML(htmlContent)
Text Content
text := extractTextFromHTML(htmlContent)
Link Discovery
links := extractLinksFromHTML(htmlContent, baseURL)
Image URLs
images := extractImagesFromHTML(htmlContent, baseURL)
Performance Tuning
Worker Configuration
crawler:
workers: 10 # Number of concurrent workers
timeout: 30s # Request timeout
maxRetries: 3 # Maximum retry attempts
retryBackoffMs: 1000 # Initial backoff time
Rate Limiting Settings
// Conservative: 1 request per 10 seconds
limiter := NewHostRateLimiter(0.1, 1)
// Moderate: 1 request per 2 seconds
limiter := NewHostRateLimiter(0.5, 2)
// Aggressive: 2 requests per second
limiter := NewHostRateLimiter(2.0, 5)
Circuit Breaker Tuning
// Sensitive: Fail fast on errors
breaker := New(Config{
FailureThreshold: 3,
RecoveryTimeout: 30*time.Second,
SuccessThreshold: 2,
})
// Tolerant: Allow more failures
breaker := New(Config{
FailureThreshold: 10,
RecoveryTimeout: 120*time.Second,
SuccessThreshold: 5,
})
Monitoring
Metrics Collected
- Successful vs failed scrapes
- Response times per domain
- Circuit breaker state changes
- Rate limiting events
- Content extraction statistics
Event Sourcing (Kafka Mode)
Complete audit trail including:
url_queued: URL added to processing queueurl_started: Scraping attempt startedurl_completed: Successfully scrapedurl_failed: Scraping failedurl_retried: Retry attempt madecircuit_open: Circuit breaker opened for domain
Best Practices
Ethical Scraping
- Always check robots.txt compliance
- Use reasonable rate limits (< 1 req/sec per domain)
- Set appropriate User-Agent header
- Implement proper retry backoff
- Respect HTTP status codes (429, 503)
Performance
- Tune worker count based on target sites
- Use circuit breakers for unreliable domains
- Implement connection pooling
- Monitor and adjust rate limits
- Use proxy rotation for large-scale scraping
Error Handling
- Distinguish between retryable and permanent errors
- Use exponential backoff for retries
- Implement dead letter queues for failed URLs
- Log detailed error information for debugging
- Set appropriate timeouts for requests
Documentation
¶
Index ¶
- type CircuitBreaker
- type Crawler
- type HostRateLimiter
- type KafkaEnhancedCrawler
- func (kec *KafkaEnhancedCrawler) BatchEnqueueURLs(ctx context.Context, urls []string, priority string, source string) error
- func (kec *KafkaEnhancedCrawler) Close() error
- func (kec *KafkaEnhancedCrawler) EnqueueURL(ctx context.Context, url string) error
- func (kec *KafkaEnhancedCrawler) EnqueueURLWithPriority(ctx context.Context, url string, priority string, source string) error
- func (kec *KafkaEnhancedCrawler) StartEnhanced(ctx context.Context)
- type RobotsCache
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type CircuitBreaker ¶
type CircuitBreaker struct {
// contains filtered or unexported fields
}
CircuitBreaker implements the circuit breaker pattern for hosts
func NewCircuitBreaker ¶
func NewCircuitBreaker( failureThreshold float64, resetTimeout time.Duration, succRequiredToClose int, rollingWindowSize int, hostErrorExpiry time.Duration, ) *CircuitBreaker
NewCircuitBreaker creates a new circuit breaker
func (*CircuitBreaker) GetState ¶
func (cb *CircuitBreaker) GetState(host string) string
GetState returns the current state of the circuit for a host
func (*CircuitBreaker) IsAllowed ¶
func (cb *CircuitBreaker) IsAllowed(host string) bool
IsAllowed checks if requests are allowed for the host
func (*CircuitBreaker) RecordFailure ¶
func (cb *CircuitBreaker) RecordFailure(host string)
RecordFailure records a failed request to the host
func (*CircuitBreaker) RecordSuccess ¶
func (cb *CircuitBreaker) RecordSuccess(host string)
RecordSuccess records a successful request to the host
func (*CircuitBreaker) Reset ¶
func (cb *CircuitBreaker) Reset(host string)
Reset resets the circuit for a host to closed state
type Crawler ¶
type Crawler struct {
// contains filtered or unexported fields
}
Crawler manages the crawling process
func NewCrawler ¶
func NewCrawler(cfg *config.Config, q queue.Queue, s database.Storage, m *metrics.MetricsCollector, p *proxy.Manager) (*Crawler, error)
NewCrawler creates a new Crawler instance
func (*Crawler) EnqueueURL ¶
EnqueueURL adds a URL to the queue for crawling
type HostRateLimiter ¶
type HostRateLimiter struct {
// contains filtered or unexported fields
}
HostRateLimiter manages rate limits for different hosts
func NewHostRateLimiter ¶
func NewHostRateLimiter(defaultQPS float64, defaultRPS int) *HostRateLimiter
NewHostRateLimiter creates a new rate limiter for hosts defaultQPS is requests per second (e.g., 0.2 for one request per 5 seconds) defaultRPS is burst capacity (max requests allowed at once)
func (*HostRateLimiter) Allow ¶
func (h *HostRateLimiter) Allow(host string) bool
Allow reports whether an event may happen for the host Does not block, but rather reports if rate limit would allow
type KafkaEnhancedCrawler ¶
type KafkaEnhancedCrawler struct {
*Crawler // Embed the basic crawler
// contains filtered or unexported fields
}
KafkaEnhancedCrawler extends the basic crawler with Kafka event sourcing and advanced features
func NewKafkaEnhancedCrawler ¶
func NewKafkaEnhancedCrawler(cfg *config.Config, q queue.Queue, s database.Storage, m *metrics.MetricsCollector, p *proxy.Manager) (*KafkaEnhancedCrawler, error)
NewKafkaEnhancedCrawler creates a new Kafka-enhanced crawler
func (*KafkaEnhancedCrawler) BatchEnqueueURLs ¶
func (kec *KafkaEnhancedCrawler) BatchEnqueueURLs(ctx context.Context, urls []string, priority string, source string) error
BatchEnqueueURLs enqueues multiple URLs efficiently using Kafka batch publishing
func (*KafkaEnhancedCrawler) Close ¶
func (kec *KafkaEnhancedCrawler) Close() error
Close closes the enhanced crawler and Kafka producer
func (*KafkaEnhancedCrawler) EnqueueURL ¶
func (kec *KafkaEnhancedCrawler) EnqueueURL(ctx context.Context, url string) error
EnqueueURL enqueues a URL with priority support and event tracking
func (*KafkaEnhancedCrawler) EnqueueURLWithPriority ¶
func (kec *KafkaEnhancedCrawler) EnqueueURLWithPriority(ctx context.Context, url string, priority string, source string) error
EnqueueURLWithPriority enqueues a URL with specific priority and source tracking
func (*KafkaEnhancedCrawler) StartEnhanced ¶
func (kec *KafkaEnhancedCrawler) StartEnhanced(ctx context.Context)
StartEnhanced starts the enhanced crawler with Kafka features
type RobotsCache ¶
type RobotsCache struct {
// contains filtered or unexported fields
}
RobotsCache caches robots.txt files and provides access control methods
func NewRobotsCache ¶
func NewRobotsCache(userAgent string, client *http.Client) *RobotsCache
NewRobotsCache creates a new robots.txt cache with the given user agent
Source Files