Skip to content
Open
Show file tree
Hide file tree
Changes from 4 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
8 changes: 6 additions & 2 deletions agent/backend/pktvisor/policy.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"bytes"
"fmt"
"net/http"
"strings"

"github.com/orb-community/orb/agent/policies"
"go.uber.org/zap"
Expand Down Expand Up @@ -60,8 +61,11 @@ func (p *pktvisorBackend) RemovePolicy(data policies.PolicyData) error {
} else {
name = data.Name
}
err := p.request(fmt.Sprintf("policies/%s", name), &resp, http.MethodDelete, http.NoBody, "application/json", RemovePolicyTimeout)
if err != nil {
if err := p.request(fmt.Sprintf("policies/%s", name), &resp, http.MethodDelete, http.NoBody, "application/json", RemovePolicyTimeout); err != nil {
if strings.Contains(err.Error(), "404") {
p.logger.Warn("ignoring error from removing a policy not found", zap.String("policy_id", data.ID), zap.String("policy_name", name))

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we change warning message to policy not found?

return nil
}
return err
}
return nil
Expand Down
7 changes: 6 additions & 1 deletion agent/comms.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"context"
"crypto/tls"
"fmt"
"strings"
"time"

mqtt "github.com/eclipse/paho.mqtt.golang"
Expand All @@ -27,8 +28,12 @@ func (a *orbAgent) connect(ctx context.Context, config config.MQTTConfig) (mqtt.
})
opts.SetConnectionLostHandler(func(client mqtt.Client, err error) {
a.logger.Error("connection to mqtt lost", zap.Error(err))
// If it is a bug on the mqttclient, stop the agent
if strings.Contains(err.Error(), "BUG") {
a.Stop(ctx)
}
a.logger.Info("reconnecting....")
client.Connect()
a.requestReconnection(ctx, a.client, config)
Comment thread
lpegoraro marked this conversation as resolved.
})
opts.SetPingTimeout(5 * time.Second)
opts.SetAutoReconnect(false)
Expand Down
7 changes: 4 additions & 3 deletions agent/otel/otlpmqttexporter/factory.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"fmt"
"go.opentelemetry.io/otel/sdk/metric"
"go.opentelemetry.io/otel/trace/noop"

"github.com/orb-community/orb/agent/otel"
"go.uber.org/zap"
Expand All @@ -13,7 +14,6 @@ import (
"go.opentelemetry.io/collector/consumer"
"go.opentelemetry.io/collector/exporter"
"go.opentelemetry.io/collector/exporter/exporterhelper"
"go.opentelemetry.io/otel/trace"
)

const (
Expand Down Expand Up @@ -55,7 +55,7 @@ func CreateDefaultSettings(logger *zap.Logger) exporter.CreateSettings {
return exporter.CreateSettings{
TelemetrySettings: component.TelemetrySettings{
Logger: logger,
TracerProvider: trace.NewNoopTracerProvider(),
TracerProvider: noop.NewTracerProvider(),
MeterProvider: metric.NewMeterProvider(),
},
BuildInfo: component.NewDefaultBuildInfo(),
Expand Down Expand Up @@ -134,7 +134,8 @@ func CreateMetricsExporter(
// explicitly disable since we rely on http.Client timeout logic.
exporterhelper.WithTimeout(exporterhelper.TimeoutSettings{Timeout: 0}),
exporterhelper.WithRetry(oCfg.RetrySettings),
exporterhelper.WithQueue(oCfg.QueueSettings))
exporterhelper.WithQueue(oCfg.QueueSettings),
exporterhelper.WithShutdown(oce.shutdown))
}

func CreateLogsExporter(
Expand Down
7 changes: 7 additions & 0 deletions agent/otel/otlpmqttexporter/otlp.go
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,13 @@ func (e *baseExporter) injectScopeLogsAttribute(logsScope plog.ScopeLogs, attrib
return logsScope
}

func (e *baseExporter) shutdown(_ context.Context) error {
if e.config.Client == nil || !(*e.config.Client).IsConnected() {
(*e.config.Client).Disconnect(0)
Comment thread
lpegoraro marked this conversation as resolved.
Outdated
}
return nil
}

func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
tr := plogotlp.NewExportRequest()
ref := tr.Logs().ResourceLogs().AppendEmpty()
Expand Down