Update MinIO bucket configuration and enhance dashboard service caching
continuous-integration/drone/push Build is passing
continuous-integration/drone/push Build is passing
- Changed the default MinIO bucket name from "opplens-documents" to "opplens" across multiple configuration files. - Introduced caching for dashboard statistics in the service layer to improve performance and reduce redundant data fetching. - Implemented a mutex for thread-safe access to cached statistics, ensuring data integrity during concurrent requests. This update streamlines the configuration for the AI summarizer and optimizes the dashboard service, enhancing overall system efficiency.
This commit is contained in:
@@ -76,7 +76,7 @@ type AISummarizerConfig struct {
|
|||||||
MinioSecretKey string `env:"AI_SUMMARIZER_MINIO_SECRET_KEY" envDefault:""`
|
MinioSecretKey string `env:"AI_SUMMARIZER_MINIO_SECRET_KEY" envDefault:""`
|
||||||
MinioUseSSL bool `env:"AI_SUMMARIZER_MINIO_USE_SSL" envDefault:"false"`
|
MinioUseSSL bool `env:"AI_SUMMARIZER_MINIO_USE_SSL" envDefault:"false"`
|
||||||
MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"`
|
MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"`
|
||||||
MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens-documents"`
|
MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// GoRulesConfig holds configuration for the GoRules status engine.
|
// GoRulesConfig holds configuration for the GoRules status engine.
|
||||||
|
|||||||
@@ -48,5 +48,5 @@ type AISummarizerConfig struct {
|
|||||||
MinioSecretKey string `env:"AI_SUMMARIZER_MINIO_SECRET_KEY" envDefault:""`
|
MinioSecretKey string `env:"AI_SUMMARIZER_MINIO_SECRET_KEY" envDefault:""`
|
||||||
MinioUseSSL bool `env:"AI_SUMMARIZER_MINIO_USE_SSL" envDefault:"false"`
|
MinioUseSSL bool `env:"AI_SUMMARIZER_MINIO_USE_SSL" envDefault:"false"`
|
||||||
MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"`
|
MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"`
|
||||||
MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens-documents"`
|
MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens"`
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
"tm/pkg/logger"
|
"tm/pkg/logger"
|
||||||
)
|
)
|
||||||
@@ -15,6 +16,7 @@ const (
|
|||||||
defaultCountriesLimit = 6
|
defaultCountriesLimit = 6
|
||||||
defaultListLimit = 5
|
defaultListLimit = 5
|
||||||
maxListLimit = 20
|
maxListLimit = 20
|
||||||
|
statisticsCacheTTL = 60 * time.Second
|
||||||
)
|
)
|
||||||
|
|
||||||
// Service defines dashboard business operations.
|
// Service defines dashboard business operations.
|
||||||
@@ -28,14 +30,25 @@ type Service interface {
|
|||||||
Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error)
|
Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type statisticsCacheEntry struct {
|
||||||
|
expiresAt time.Time
|
||||||
|
value *StatisticsReportResponse
|
||||||
|
}
|
||||||
|
|
||||||
type service struct {
|
type service struct {
|
||||||
repo Repository
|
repo Repository
|
||||||
logger logger.Logger
|
logger logger.Logger
|
||||||
|
statisticsMu sync.Mutex
|
||||||
|
statistics map[int]statisticsCacheEntry
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewService creates a dashboard service.
|
// NewService creates a dashboard service.
|
||||||
func NewService(repo Repository, log logger.Logger) Service {
|
func NewService(repo Repository, log logger.Logger) Service {
|
||||||
return &service{repo: repo, logger: log}
|
return &service{
|
||||||
|
repo: repo,
|
||||||
|
logger: log,
|
||||||
|
statistics: make(map[int]statisticsCacheEntry),
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *service) Summary(ctx context.Context, query SummaryQuery) (*SummaryResponse, error) {
|
func (s *service) Summary(ctx context.Context, query SummaryQuery) (*SummaryResponse, error) {
|
||||||
@@ -160,6 +173,10 @@ func (s *service) Recent(ctx context.Context, query RecentQuery) (*RecentRespons
|
|||||||
|
|
||||||
func (s *service) Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error) {
|
func (s *service) Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error) {
|
||||||
days := trendDays(query.Days)
|
days := trendDays(query.Days)
|
||||||
|
if cached, ok := s.cachedStatistics(days); ok {
|
||||||
|
return cached, nil
|
||||||
|
}
|
||||||
|
|
||||||
startUnix := trendStartUnix(days)
|
startUnix := trendStartUnix(days)
|
||||||
|
|
||||||
s.logger.Info("Fetching dashboard statistics report", map[string]interface{}{
|
s.logger.Info("Fetching dashboard statistics report", map[string]interface{}{
|
||||||
@@ -174,9 +191,37 @@ func (s *service) Statistics(ctx context.Context, query StatisticsQuery) (*Stati
|
|||||||
return nil, fmt.Errorf("dashboard statistics: %w", err)
|
return nil, fmt.Errorf("dashboard statistics: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
s.storeStatistics(days, out)
|
||||||
return out, nil
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *service) cachedStatistics(days int) (*StatisticsReportResponse, bool) {
|
||||||
|
now := time.Now()
|
||||||
|
|
||||||
|
s.statisticsMu.Lock()
|
||||||
|
defer s.statisticsMu.Unlock()
|
||||||
|
|
||||||
|
entry, ok := s.statistics[days]
|
||||||
|
if !ok || now.After(entry.expiresAt) {
|
||||||
|
if ok {
|
||||||
|
delete(s.statistics, days)
|
||||||
|
}
|
||||||
|
return nil, false
|
||||||
|
}
|
||||||
|
|
||||||
|
return entry.value, true
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *service) storeStatistics(days int, value *StatisticsReportResponse) {
|
||||||
|
s.statisticsMu.Lock()
|
||||||
|
defer s.statisticsMu.Unlock()
|
||||||
|
|
||||||
|
s.statistics[days] = statisticsCacheEntry{
|
||||||
|
expiresAt: time.Now().Add(statisticsCacheTTL),
|
||||||
|
value: value,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func closingWindowHours(hours int) int {
|
func closingWindowHours(hours int) int {
|
||||||
if hours <= 0 {
|
if hours <= 0 {
|
||||||
return defaultClosingWindowHours
|
return defaultClosingWindowHours
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import (
|
|||||||
|
|
||||||
"go.mongodb.org/mongo-driver/v2/bson"
|
"go.mongodb.org/mongo-driver/v2/bson"
|
||||||
mongodriver "go.mongodb.org/mongo-driver/v2/mongo"
|
mongodriver "go.mongodb.org/mongo-driver/v2/mongo"
|
||||||
|
"golang.org/x/sync/errgroup"
|
||||||
)
|
)
|
||||||
|
|
||||||
const noticesCollectionName = "notices"
|
const noticesCollectionName = "notices"
|
||||||
@@ -19,32 +20,66 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64)
|
|||||||
endDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
|
endDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
|
||||||
startDay := endDay.AddDate(0, 0, -(days - 1))
|
startDay := endDay.AddDate(0, 0, -(days - 1))
|
||||||
|
|
||||||
scrapedTED, err := r.dailyCountByTimestamp(ctx, noticesCollectionName, bson.M{
|
var (
|
||||||
|
scrapedTED map[string]int64
|
||||||
|
scrapedDocuments map[string]int64
|
||||||
|
translatedNotices map[string]int64
|
||||||
|
totalDocuments int64
|
||||||
|
totalTranslated int64
|
||||||
|
)
|
||||||
|
|
||||||
|
g, gctx := errgroup.WithContext(ctx)
|
||||||
|
|
||||||
|
g.Go(func() error {
|
||||||
|
counts, err := r.dailyCountByTimestamp(gctx, noticesCollectionName, bson.M{
|
||||||
"source": notice.TenderSourceTEDScraper,
|
"source": notice.TenderSourceTEDScraper,
|
||||||
"processing_metadata.scraped_at": bson.M{"$gte": startUnix, "$gt": 0},
|
"processing_metadata.scraped_at": bson.M{"$gte": startUnix, "$gt": 0},
|
||||||
}, "processing_metadata.scraped_at")
|
}, "processing_metadata.scraped_at")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("scraped ted per day: %w", err)
|
return fmt.Errorf("scraped ted per day: %w", err)
|
||||||
}
|
}
|
||||||
|
scrapedTED = counts
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
scrapedDocuments, err := r.scrapedDocumentsPerDay(ctx, startUnix)
|
g.Go(func() error {
|
||||||
|
counts, err := r.scrapedDocumentsPerDay(gctx, startUnix)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("scraped documents per day: %w", err)
|
return fmt.Errorf("scraped documents per day: %w", err)
|
||||||
}
|
}
|
||||||
|
scrapedDocuments = counts
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
translatedNotices, err := r.translatedNoticesPerDay(ctx, startDay, endDay)
|
g.Go(func() error {
|
||||||
|
counts, err := r.translatedNoticesPerDay(gctx, startDay, endDay)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("translated notices per day: %w", err)
|
return fmt.Errorf("translated notices per day: %w", err)
|
||||||
}
|
}
|
||||||
|
translatedNotices = counts
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
totalDocuments, err := r.totalScrapedDocuments(ctx)
|
g.Go(func() error {
|
||||||
|
total, err := r.totalScrapedDocuments(gctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("total scraped documents: %w", err)
|
return fmt.Errorf("total scraped documents: %w", err)
|
||||||
}
|
}
|
||||||
|
totalDocuments = total
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
totalTranslated, err := r.metricsCounter.Get(ctx, orm.AITranslationSuccessCounterKey)
|
g.Go(func() error {
|
||||||
|
total, err := r.metricsCounter.Get(gctx, orm.AITranslationSuccessCounterKey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("total translated tenders: %w", err)
|
return fmt.Errorf("total translated tenders: %w", err)
|
||||||
|
}
|
||||||
|
totalTranslated = total
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
|
if err := g.Wait(); err != nil {
|
||||||
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
return &StatisticsReportResponse{
|
return &StatisticsReportResponse{
|
||||||
@@ -94,6 +129,10 @@ func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64
|
|||||||
{{Key: "$match", Value: bson.M{
|
{{Key: "$match", Value: bson.M{
|
||||||
"processing_metadata.documents_scraped": true,
|
"processing_metadata.documents_scraped": true,
|
||||||
"scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}},
|
"scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}},
|
||||||
|
"$or": []bson.M{
|
||||||
|
{"processing_metadata.documents_scraped_at": bson.M{"$gte": startUnix}},
|
||||||
|
{"scraped_documents.scraped_at": bson.M{"$gte": startUnix}},
|
||||||
|
},
|
||||||
}}},
|
}}},
|
||||||
{{Key: "$unwind", Value: "$scraped_documents"}},
|
{{Key: "$unwind", Value: "$scraped_documents"}},
|
||||||
{{Key: "$addFields", Value: bson.M{
|
{{Key: "$addFields", Value: bson.M{
|
||||||
|
|||||||
@@ -39,6 +39,10 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Re
|
|||||||
noticeIndexes := []orm.Index{
|
noticeIndexes := []orm.Index{
|
||||||
// processing_metadata.processed
|
// processing_metadata.processed
|
||||||
*orm.NewIndex("processing_metadata_processed_idx", bson.D{{Key: "processing_metadata.processed", Value: 1}}),
|
*orm.NewIndex("processing_metadata_processed_idx", bson.D{{Key: "processing_metadata.processed", Value: 1}}),
|
||||||
|
*orm.NewIndex("source_scraped_at_idx", bson.D{
|
||||||
|
{Key: "source", Value: 1},
|
||||||
|
{Key: "processing_metadata.scraped_at", Value: 1},
|
||||||
|
}),
|
||||||
*orm.NewIndex("main_classification_idx", bson.D{{Key: "main_classification", Value: 1}}),
|
*orm.NewIndex("main_classification_idx", bson.D{{Key: "main_classification", Value: 1}}),
|
||||||
*orm.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}),
|
*orm.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}),
|
||||||
// One row per TED ContractNoticeID when present (parallel scrape races).
|
// One row per TED ContractNoticeID when present (parallel scrape races).
|
||||||
|
|||||||
@@ -150,6 +150,13 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Te
|
|||||||
{Key: "tender_deadline", Value: 1},
|
{Key: "tender_deadline", Value: 1},
|
||||||
{Key: "created_at", Value: 1},
|
{Key: "created_at", Value: 1},
|
||||||
}),
|
}),
|
||||||
|
*orm.NewIndex("documents_scraped_at_idx", bson.D{
|
||||||
|
{Key: "processing_metadata.documents_scraped", Value: 1},
|
||||||
|
{Key: "processing_metadata.documents_scraped_at", Value: -1},
|
||||||
|
}),
|
||||||
|
*orm.NewIndex("scraped_documents_scraped_at_idx", bson.D{
|
||||||
|
{Key: "scraped_documents.scraped_at", Value: 1},
|
||||||
|
}),
|
||||||
// Index for CPV matching queries
|
// Index for CPV matching queries
|
||||||
*orm.NewIndex("cpv_matching_idx", bson.D{
|
*orm.NewIndex("cpv_matching_idx", bson.D{
|
||||||
{Key: "status", Value: 1},
|
{Key: "status", Value: 1},
|
||||||
|
|||||||
@@ -26,7 +26,7 @@ type Config struct {
|
|||||||
MinioSecretKey string // MinIO secret access key
|
MinioSecretKey string // MinIO secret access key
|
||||||
MinioUseSSL bool // Whether to use SSL for MinIO connection
|
MinioUseSSL bool // Whether to use SSL for MinIO connection
|
||||||
MinioRegion string // MinIO region
|
MinioRegion string // MinIO region
|
||||||
MinioBucket string // MinIO bucket name (default: "opplens-documents")
|
MinioBucket string // MinIO bucket name (default: "opplens")
|
||||||
MinioTimeout time.Duration // Timeout for MinIO operations
|
MinioTimeout time.Duration // Timeout for MinIO operations
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -37,7 +37,7 @@ func DefaultConfig() *Config {
|
|||||||
APIRetryCount: 2,
|
APIRetryCount: 2,
|
||||||
APIRetryDelay: 3 * time.Second,
|
APIRetryDelay: 3 * time.Second,
|
||||||
|
|
||||||
MinioBucket: "opplens-documents",
|
MinioBucket: "opplens",
|
||||||
MinioRegion: "us-east-1",
|
MinioRegion: "us-east-1",
|
||||||
MinioUseSSL: false,
|
MinioUseSSL: false,
|
||||||
MinioTimeout: 30 * time.Second,
|
MinioTimeout: 30 * time.Second,
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// procedurePrefix is prepended to every contract folder id in the MinIO key.
|
// procedurePrefix is prepended to every contract folder id in the MinIO key.
|
||||||
// The AI team's storage layout in bucket opplens-documents is:
|
// The AI team's storage layout in bucket opplens is:
|
||||||
//
|
//
|
||||||
// PROC_<contractFolderID>/<noticePublicationID>/tender.json
|
// PROC_<contractFolderID>/<noticePublicationID>/tender.json
|
||||||
// PROC_<contractFolderID>/<noticePublicationID>/documents/{files}
|
// PROC_<contractFolderID>/<noticePublicationID>/documents/{files}
|
||||||
|
|||||||
+39
-6
@@ -104,14 +104,47 @@ func (c *Counter) GetDailyCounts(ctx context.Context, startDay, endDay time.Time
|
|||||||
return map[string]int64{}, nil
|
return map[string]int64{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
counts := make(map[string]int64)
|
keys := make([]string, 0)
|
||||||
|
dates := make([]string, 0)
|
||||||
for day := startDay; !day.After(endDay); day = day.AddDate(0, 0, 1) {
|
for day := startDay; !day.After(endDay); day = day.AddDate(0, 0, 1) {
|
||||||
date := day.Format("2006-01-02")
|
dates = append(dates, day.Format("2006-01-02"))
|
||||||
value, err := c.Get(ctx, AITranslationSuccessDailyCounterKey(day))
|
keys = append(keys, AITranslationSuccessDailyCounterKey(day))
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
}
|
||||||
counts[date] = value
|
|
||||||
|
counts := make(map[string]int64, len(dates))
|
||||||
|
for _, date := range dates {
|
||||||
|
counts[date] = 0
|
||||||
|
}
|
||||||
|
|
||||||
|
cursor, err := c.mongo.GetCollection(metricsCountersCollection).Find(ctx, bson.M{
|
||||||
|
"_id": bson.M{"$in": keys},
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("mongo counter get daily counts: %w", err)
|
||||||
|
}
|
||||||
|
defer cursor.Close(ctx)
|
||||||
|
|
||||||
|
keyToDate := make(map[string]string, len(keys))
|
||||||
|
for i, key := range keys {
|
||||||
|
keyToDate[key] = dates[i]
|
||||||
|
}
|
||||||
|
|
||||||
|
for cursor.Next(ctx) {
|
||||||
|
var doc struct {
|
||||||
|
ID string `bson:"_id"`
|
||||||
|
Count int64 `bson:"count"`
|
||||||
|
}
|
||||||
|
if err := cursor.Decode(&doc); err != nil {
|
||||||
|
return nil, fmt.Errorf("mongo counter decode daily count: %w", err)
|
||||||
|
}
|
||||||
|
date, ok := keyToDate[doc.ID]
|
||||||
|
if !ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
counts[date] = doc.Count
|
||||||
|
}
|
||||||
|
if err := cursor.Err(); err != nil {
|
||||||
|
return nil, fmt.Errorf("mongo counter iterate daily counts: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return counts, nil
|
return counts, nil
|
||||||
|
|||||||
Reference in New Issue
Block a user