Skip to content

fix: alienvault multipage disk read + rate limit skip #67

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Runaho merged 1 commit into from
May 28, 2026
Merged

fix: alienvault multipage disk read + rate limit skip #67

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
127 changes: 100 additions & 27 deletions features/providers/alienvault/provider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"encoding/json"
"fmt"
"io"
"os"
"strings"
"time"

Expand Down Expand Up @@ -127,7 +128,7 @@ func (p *alienvaultProvider) Fetch() (io.Reader, error) {
pageCount := 0
totalIndicators := 0

for {
for {
// Apply rate limiting between requests
if pageCount > 0 {
time.Sleep(p.rateLimit)
Expand Down Expand Up @@ -346,30 +347,76 @@ func (p *alienvaultProvider) FetchPages(ctx context.Context) (<-chan base.PagePa
pageCount := 0
totalIndicators := 0

// If resuming, fast-forward through local pages by advancing URL
// and skipping already-saved pages. We rely on the next_url from
// each saved page's meta entry to reconstruct the URL chain.
// If resuming, read pages from disk and parse them, then continue from network.
// Skip rate limit for disk reads. Use stored next_url to resume correctly.
if !freshRun {
// Load meta to find the URL for the next page to fetch
meta, err := utils.GetPageMetadata(storePath, providerName)
if err == nil && meta != nil && len(meta.Pages) >= startPage-1 {
// We have pages 1..startPage-1. The next page is startPage.
// But we don't store the next_url per-page — the original Fetch()
// would have gotten it from the API. So we cannot truly resume
// the remote fetch from a partial local state without re-fetching
// pages 1..startPage-1 to get their next_url.
//
// Simplest correct approach: start fresh but skip saving pages 1..startPage-1
// since they already exist on disk. This costs one API call per already-saved
// page but gives us correct next_url chain.
log.Info().
Int("existing_pages", startPage-1).
Int("next_page", startPage).
Msg("Skipping existing local pages, fetching from API to get correct next_url chain")
if err != nil || meta == nil || len(meta.Pages) < startPage-1 {
log.Warn().
Err(err).
Int("start_page", startPage).
Int("meta_pages", func() int { if meta == nil { return 0 }; return len(meta.Pages) }()).
Msg("Cannot read meta or not enough pages — falling back to network fetch from start")
startPage = 1
currentURL = p.SourceURL
} else {
// Reconstruct URL chain from stored next_page_urls.
// Read pages 1..startPage-1 from disk, parse and submit entries.
for i := 1; i < startPage; i++ {
pageInfo := meta.Pages[i-1]
pagePath := storePath + "/" + providerName + "/" + pageInfo.File
data, err := os.ReadFile(pagePath)
if err != nil {
log.Warn().Err(err).Int("page", i).Msg("Failed to read page from disk — skipping")
continue
}

var response OTXResponse
if err := json.Unmarshal(data, &response); err != nil {
log.Warn().Err(err).Int("page", i).Msg("Failed to unmarshal page from disk — skipping")
continue
}

pageIndicators := 0
processID := ""
if p.ProcessID != nil {
processID = p.ProcessID.String()
}
for _, pulse := range response.Results {
for _, indicator := range pulse.Indicators {
entry, err := indicatorToEntry(&indicator, providerName, processID)
if err != nil || entry == nil {
continue
}
collector.Submit(entry)
pageIndicators++
}
}
pageCount++
totalIndicators += pageIndicators

log.Info().
Int("page", i).
Int("indicators", pageIndicators).
Str("next_url", pageInfo.NextPageURL).
Msg("Processed page from disk")
}

// Set currentURL for the next network fetch.
// Use stored next_page_url from the last disk page if available,
// otherwise fall back to SourceURL (network will detect if more pages exist).
lastPageInfo := meta.Pages[startPage-2]
if lastPageInfo.NextPageURL != "" {
currentURL = lastPageInfo.NextPageURL
log.Info().Str("resume_url", currentURL).Msg("Resuming from stored next_page_url")
} else {
// Disk page has no next_url — it may be the last page of a previous run,
// or the previous run was interrupted before the next_url was stored.
// Resume from SourceURL and let the API decide if there are more pages.
currentURL = p.SourceURL
log.Info().Msg("No stored next_page_url — resuming from SourceURL")
}
}
// Reset to first page to rebuild URL chain correctly
startPage = 1
currentURL = p.SourceURL
}

for {
Expand All @@ -381,7 +428,8 @@ func (p *alienvaultProvider) FetchPages(ctx context.Context) (<-chan base.PagePa
default:
}

// Apply rate limiting between requests
// Apply rate limiting between network requests only (pageCount already
// accounts for disk pages when resuming).
if pageCount > 0 {
time.Sleep(p.rateLimit)
}
Expand Down Expand Up @@ -511,9 +559,10 @@ func (p *alienvaultProvider) FetchPages(ctx context.Context) (<-chan base.PagePa
return
}

// Save page to disk immediately (per-page persistence)
// Save page to disk immediately (per-page persistence), including next_page_url
fetchedAt := time.Now()
if _, err := utils.SavePageData(storePath, providerName, pageCount, body, 0, fetchedAt); err != nil {
nextPageURL := response.Next
if _, err := utils.SavePageData(storePath, providerName, pageCount, body, 0, fetchedAt, nextPageURL); err != nil {
log.Warn().Err(err).Int("page", pageCount).Msg("Failed to save page to disk — continuing anyway")
}

Expand All @@ -537,11 +586,35 @@ func (p *alienvaultProvider) FetchPages(ctx context.Context) (<-chan base.PagePa

totalIndicators += pageIndicators

// Determine next page URL
// Determine if there is a next page.
// response.Next == "" means no more pages. But if currentURL != p.SourceURL
// (i.e., we resumed from a stored next_url), empty response.Next means
// the previous fetch was interrupted mid-page — we should NOT treat this as done.
hasNext := response.Next != "" && response.Next != currentURL
nextURL := response.Next

if !hasNext {
// If we are on a resumed URL (not the first page) and the response has no next_url,
// this page was already fetched — the previous run stored it with next_url="".
// It is not a new page, so we are done.
if pageCount >= startPage && currentURL != p.SourceURL {
log.Info().
Int("pages_fetched", pageCount).
Int("total_indicators", totalIndicators).
Int("bytes", len(body)).
Msg("Multi-page fetch complete")
resultChan <- base.PageParseResult{
PageNumber: pageCount,
Indicators: pageIndicators,
Bytes: int64(len(body)),
FetchedAt: fetchedAt,
HasNextPage: false,
NextPageURL: "",
Entries: nil,
}
return
}
// Fresh run or first page: no more pages from API
log.Info().
Int("pages_fetched", pageCount).
Int("total_indicators", totalIndicators).
Expand Down Expand Up @@ -686,4 +759,4 @@ func indicatorToEntry(indicator *OTXIndicator, source, processID string) (*entri
Msg("unsupported indicator type — skipping")
return nil, nil
}
}
}
18 changes: 11 additions & 7 deletions internal/utils/multipage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,9 +35,10 @@ type PageMetadata struct {

// PageInfo describes a single page file on disk.
type PageInfo struct {
File string `json:"file"`
FetchedAt time.Time `json:"fetched_at"`
Indicators int `json:"indicators"`
File string `json:"file"`
FetchedAt time.Time `json:"fetched_at"`
Indicators int `json:"indicators"`
NextPageURL string `json:"next_page_url,omitempty"`
}

// --- Per-page persistence functions ---
Expand All @@ -54,7 +55,8 @@ func GetProviderDataDir(storePath, providerName string) (string, error) {

// SavePageData saves a single page's raw response to disk and updates the meta file.
// pageNum is 1-indexed. indicatorCount may be 0 if not yet parsed.
func SavePageData(storePath, providerName string, pageNum int, data []byte, indicatorCount int, fetchedAt time.Time) (string, error) {
// nextPageURL is stored in metadata to enable resume without re-fetching.
func SavePageData(storePath, providerName string, pageNum int, data []byte, indicatorCount int, fetchedAt time.Time, nextPageURL string) (string, error) {
dir, err := GetProviderDataDir(storePath, providerName)
if err != nil {
return "", err
Expand Down Expand Up @@ -85,9 +87,10 @@ func SavePageData(storePath, providerName string, pageNum int, data []byte, indi
meta.Pages = append(meta.Pages, PageInfo{})
}
meta.Pages[pageNum-1] = PageInfo{
File: pageFilename,
FetchedAt: fetchedAt,
Indicators: indicatorCount,
File: pageFilename,
FetchedAt: fetchedAt,
Indicators: indicatorCount,
NextPageURL: nextPageURL,
}
if pageNum > meta.TotalPages {
meta.TotalPages = pageNum
Expand All @@ -102,6 +105,7 @@ func SavePageData(storePath, providerName string, pageNum int, data []byte, indi
Int("page", pageNum).
Str("file", pageFilename).
Int("bytes", len(data)).
Str("next_page_url", nextPageURL).
Msg("Page saved to disk")

return pagePath, nil
Expand Down
Loading