Skip to content
Draft
Show file tree
Hide file tree
Changes from 8 commits
Commits
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
1 change: 1 addition & 0 deletions .golangci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,7 @@ linters:
- forcetypeassert # for tests' brevity sake
- funlen # tests may be long
- gocognit # triggered by subtests
- gosec # not critical, too many things to disable
- ireturn # we have exceptions, so need to silence them in tests
- lll # tests often require long lines
- maintidx # not critical for tests
Expand Down
10 changes: 8 additions & 2 deletions admin/commands/summary.go
Original file line number Diff line number Diff line change
Expand Up @@ -189,7 +189,10 @@ func addServerData(ctx context.Context, zipW *zip.Writer, usePprof bool) {

addData(zipW, path.Join("server", rf.Name), rf.Modified, rc) //nolint:gosec

rc.Close() //nolint:errcheck
err = rc.Close()
if err != nil {
logrus.Errorf("Failed to close zip archive %s: %v", rf.Name, err)
}
}
}

Expand Down Expand Up @@ -277,7 +280,10 @@ func downloadFile(ctx context.Context, zipW *zip.Writer, url, fileName string) e
}
addData(zipW, path.Join(fileName, rf.Name), rf.Modified, rc) //nolint:gosec

rc.Close() //nolint:errcheck
err = rc.Close()
if err != nil {
logrus.Errorf("Failed to close zip archive %s: %v", rf.Name, err)
}
}
return nil
}
Expand Down
5 changes: 4 additions & 1 deletion agent/agents/mongodb/mongolog/mongodb.go
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,10 @@ func (m *MongoDB) Run(ctx context.Context) {
var log Mongolog

defer func() {
log.Stop() //nolint:errcheck
err := log.Stop()
if err != nil {
m.l.Errorf("Can't stop mongolog, reason: %v", err)
}
log = nil
m.changes <- agents.Change{Status: inventoryv1.AgentStatus_AGENT_STATUS_DONE}
close(m.changes)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@
}

// Start starts but doesn't wait until it exits.
func (c *Collector) Start(context.Context) (<-chan proto.SystemProfile, error) {
func (c *Collector) Start(ctx context.Context) (<-chan proto.SystemProfile, error) {
c.m.Lock()
defer c.m.Unlock()
if c.running {
Expand All @@ -86,7 +86,6 @@
ready.L.Lock()
defer ready.L.Unlock()

ctx := context.Background()
labels := pprof.Labels("component", "mongodb.aggregator")
go pprof.Do(ctx, labels, func(ctx context.Context) {
start(
Expand Down Expand Up @@ -175,15 +174,15 @@
logger.Traceln("connect and collect is called")
query := createQuery(dbName, startTime)

timeoutCtx, cancel := context.WithTimeout(context.TODO(), cursorTimeout)
timeoutCtx, cancel := context.WithTimeout(ctx, cursorTimeout)
defer cancel()
cursor, err := createIterator(timeoutCtx, collection, query)
if err != nil {
logger.Errorf("couldn't create system.profile iterator, reason: %v", err)
return
}
// do not cancel cursor closing when ctx is canceled
// Ensure cursor is closed even if parent context is canceled to prevent resource leaks.
defer cursor.Close(context.Background()) //nolint:errcheck

Check failure on line 185 in agent/agents/mongodb/profiler/internal/collector/collector.go

View workflow job for this annotation

GitHub Actions / Checks

Non-inherited new context, use function like `context.WithXXX` instead (contextcheck)

// we got iterator, we are ready
signalReady(ready)
Expand All @@ -204,7 +203,7 @@
}()

for {
for cursor.TryNext(context.TODO()) {
for cursor.TryNext(ctx) {
doc := proto.SystemProfile{}
e := cursor.Decode(&doc)
if e != nil {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,45 @@ func TestCollector(t *testing.T) {
assert.Len(t, profiles, maxDocs*maxLoops)
}

func TestCollectorContextCancel(t *testing.T) {
url := "mongodb://root:root-password@127.0.0.1:27017"
client, err := createSession(url, "pmm-agent-test")
require.NoError(t, err)

// Create a sub-context specifically for this test to trigger cancellation.
ctx, cancel := context.WithCancel(t.Context())

ctr := New(client, "test_context_cancel", logrus.WithField("component", "collector-test"))

docsChan, err := ctr.Start(ctx)
require.NoError(t, err)

// Verify the collector is running.
assert.Equal(t, "collector", ctr.Name())

// Cancel the context to signal the internal goroutine to exit.
cancel()

// Use Stop() as a synchronization point. It waits for the internal WaitGroup,
// which confirms that the start() and connectAndCollect() goroutines have exited.
stopDone := make(chan struct{})
go func() {
ctr.Stop()
close(stopDone)
}()

select {
case <-stopDone:
// Success: Internal goroutines shut down gracefully.
case <-time.After(5 * time.Second):
t.Fatal("Collector did not stop via context cancellation within timeout")
}

// Verify that the data channel was closed.
_, ok := <-docsChan
assert.False(t, ok, "docsChan should be closed after collector stops")
}

func genData(ctx context.Context, client *mongo.Client, maxLoops, maxDocs int) {
interval := time.Millisecond

Expand Down
7 changes: 5 additions & 2 deletions agent/agents/mongodb/profiler/internal/parser/parser.go
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ type Parser struct {
}

// Start starts but doesn't wait until it exits.
func (p *Parser) Start(context.Context) error {
func (p *Parser) Start(ctx context.Context) error {
p.m.Lock()
defer p.m.Unlock()
if p.running {
Expand All @@ -70,7 +70,6 @@ func (p *Parser) Start(context.Context) error {
p.wg = &sync.WaitGroup{}
p.wg.Add(1)

ctx := context.Background()
labels := pprof.Labels("component", "mongodb.monitor")
go pprof.Do(ctx, labels, func(ctx context.Context) {
start(
Expand Down Expand Up @@ -126,6 +125,8 @@ func start(
// PMM-13947
case <-doneChan:
return
case <-ctx.Done():
return
default:
// just continue if not
}
Expand All @@ -149,6 +150,8 @@ func start(
// doneChan needs to be also in separate select statement
// as docsChan could be always picked since select picks channels pseudo randomly
return
case <-ctx.Done():
return
}
}
}
32 changes: 32 additions & 0 deletions agent/agents/mongodb/profiler/internal/parser/parser_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,38 @@ func TestParserStartStop(t *testing.T) {
parser1.Stop()
}

func TestParserContextCancel(t *testing.T) {
docsChan := make(chan pm.SystemProfile)
a := aggregator.New(time.Now(), "test-id", logrus.WithField("component", "aggregator"), truncate.GetMongoDBDefaultMaxQueryLength())

ctx, cancel := context.WithCancel(context.Background())
parser := New(docsChan, a, logrus.WithField("component", "test-parser"))

err := parser.Start(ctx)
require.NoError(t, err)

// Verify parser reported as running
assert.Equal(t, "parser", parser.Name())

// Cancel the context to trigger the new shutdown path in start()
cancel()

// We use Stop() as a synchronization point because it waits for the internal WaitGroup.
// If the context cancellation logic is correct, Stop() should return immediately.
done := make(chan struct{})
go func() {
parser.Stop()
close(done)
}()

select {
case <-done:
// Success: the parser stopped correctly via context cancellation
case <-time.After(1 * time.Second):
t.Fatal("Parser did not stop after context cancellation within timeout")
}
}

func TestParserRunning(t *testing.T) {
oldInterval := aggregator.DefaultInterval
aggregator.DefaultInterval = 10 * time.Second
Expand Down
5 changes: 4 additions & 1 deletion agent/agents/mongodb/profiler/internal/profiler.go
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,10 @@ func (p *Profiler) Stop() error {
p.sender.Stop()

// close the session; do it after goroutine is closed
p.client.Disconnect(context.TODO()) //nolint:errcheck
err := p.client.Disconnect(context.Background())
if err != nil {
p.logger.Errorf("Failed to disconnect client from MongoDB, reason: %v", err)
}
Comment thread
maxkondr marked this conversation as resolved.

// set state to "not running"
p.running = false
Expand Down
5 changes: 4 additions & 1 deletion agent/agents/mongodb/profiler/mongodb.go
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,10 @@ func (m *MongoDB) Run(ctx context.Context) {
var prof Profiler

defer func() {
prof.Stop() //nolint:errcheck
err := prof.Stop()
if err != nil {
m.l.Errorf("Can't stop profiler, reason: %v", err)
}
prof = nil
m.changes <- agents.Change{Status: inventoryv1.AgentStatus_AGENT_STATUS_DONE}
close(m.changes)
Expand Down
2 changes: 1 addition & 1 deletion agent/agents/mongodb/shared/aggregator/aggregator.go
Original file line number Diff line number Diff line change
Expand Up @@ -281,7 +281,7 @@ func (a *Aggregator) createResult(_ context.Context) *report.Result {
ClientHost: v.Client,
AgentId: a.agentID,
AgentType: inventoryv1.AgentType_AGENT_TYPE_QAN_MONGODB_PROFILER_AGENT,
PeriodStartUnixSecs: uint32(a.timeStart.Truncate(1 * time.Minute).Unix()),
PeriodStartUnixSecs: uint32(a.timeStart.Truncate(1 * time.Minute).Unix()), //nolint:gosec // Unix timestamp fits in uint32 until 2106
PeriodLengthSecs: uint32(a.d.Seconds()),
Example: query,
ExampleType: agentv1.ExampleType_EXAMPLE_TYPE_RANDOM,
Expand Down
7 changes: 5 additions & 2 deletions agent/agents/mysql/perfschema/perfschema.go
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,10 @@ func newPerfSchema(params *newPerfSchemaParams) (*PerfSchema, error) {
// Run extracts performance data and sends it to the channel until ctx is canceled.
func (m *PerfSchema) Run(ctx context.Context) {
defer func() {
m.dbCloser.Close() //nolint:errcheck
err := m.dbCloser.Close()
if err != nil {
m.l.WithError(err).Error("Failed to close database connection")
}
m.changes <- agents.Change{Status: inventoryv1.AgentStatus_AGENT_STATUS_DONE}
close(m.changes)
}()
Expand Down Expand Up @@ -345,7 +348,7 @@ func (m *PerfSchema) getNewBuckets(periodStart time.Time, periodLengthSecs uint3
}

buckets := makeBuckets(current, prev, m.l, m.maxQueryLength)
startS := uint32(periodStart.Unix())
startS := uint32(periodStart.Unix()) //nolint:gosec // Unix timestamp fits in uint32 until 2106
m.l.Debugf("Made %d buckets out of %d summaries in %s+%d interval.",
len(buckets), len(current), periodStart.Format("15:04:05"), periodLengthSecs)

Expand Down
2 changes: 1 addition & 1 deletion agent/agents/mysql/slowlog/slowlog.go
Original file line number Diff line number Diff line change
Expand Up @@ -454,7 +454,7 @@ func makeBuckets(
ClientHost: v.Host,
AgentId: agentID,
AgentType: inventoryv1.AgentType_AGENT_TYPE_QAN_MYSQL_SLOWLOG_AGENT,
PeriodStartUnixSecs: uint32(periodStart.Unix()),
PeriodStartUnixSecs: uint32(periodStart.Unix()), //nolint:gosec // Unix timestamp fits in uint32 until 2106
PeriodLengthSecs: periodLengthSecs,
NumQueries: float32(v.TotalQueries),
Errors: errListsToMap(v.ErrorsCode, v.ErrorsCount),
Expand Down
26 changes: 21 additions & 5 deletions agent/agents/postgres/pgstatmonitor/pgstatmonitor.go
Original file line number Diff line number Diff line change
Expand Up @@ -287,7 +287,10 @@ func getPGMonitorVersion(q *reform.Querier) (pgStatMonitorVersion, pgStatMonitor
// Run extracts stats data and sends it to the channel until ctx is canceled.
func (m *PGStatMonitorQAN) Run(ctx context.Context) {
defer func() {
m.dbCloser.Close() //nolint:errcheck
err := m.dbCloser.Close()
if err != nil {
m.l.WithError(err).Error("Failed to close DB connection")
}
m.changes <- agents.Change{Status: inventoryv1.AgentStatus_AGENT_STATUS_DONE}
close(m.changes)
}()
Expand Down Expand Up @@ -609,7 +612,7 @@ func (m *PGStatMonitorQAN) makeBuckets(current, cache map[time.Time]map[string]*
NumQueries: count,
ClientHost: currentPSM.ClientIP,
AgentType: inventoryv1.AgentType_AGENT_TYPE_QAN_POSTGRESQL_PGSTATMONITOR_AGENT,
PeriodStartUnixSecs: uint32(currentPSM.BucketStartTime.Unix()),
PeriodStartUnixSecs: uint32(currentPSM.BucketStartTime.Unix()), //nolint:gosec // Unix timestamp fits in uint32 until 2106
},
Postgresql: &agentv1.MetricsBucket_PostgreSQL{},
}
Expand Down Expand Up @@ -725,7 +728,11 @@ func (m *PGStatMonitorQAN) makeBuckets(current, cache map[time.Time]map[string]*
func parseHistogramFromRespCalls(respCalls pq.StringArray, prevRespCalls pq.StringArray, vPGSM pgStatMonitorVersion) ([]*agentv1.HistogramItem, error) {
histogram := getHistogramRangesArray(vPGSM)
for k, v := range respCalls {
val, err := strconv.ParseInt(v, 10, 32)
if k >= len(histogram) {
break
}
// Use ParseUint with bitSize 32 to ensure non-negative values that fit in uint32
val, err := strconv.ParseUint(v, 10, 32)
if err != nil {
return nil, fmt.Errorf("failed to parse histogram: %w", err)
}
Expand All @@ -734,12 +741,21 @@ func parseHistogramFromRespCalls(respCalls pq.StringArray, prevRespCalls pq.Stri
}

for k, v := range prevRespCalls {
val, err := strconv.ParseInt(v, 10, 32)
if k >= len(histogram) {
break
}
val, err := strconv.ParseUint(v, 10, 32)
if err != nil {
return nil, fmt.Errorf("failed to parse histogram: %w", err)
}

histogram[k].Frequency -= uint32(val)
uVal := uint32(val)
if histogram[k].Frequency >= uVal {
histogram[k].Frequency -= uVal
} else {
// Handle counter resets or inconsistencies by capping at 0
histogram[k].Frequency = 0
}
}

return histogram, nil
Expand Down
Loading
Loading