fix(inputs.opcua): Ensure we are connected after reconnect (#13514)
This commit is contained in:
parent
96b9845853
commit
261e0223a7
|
|
@ -18,6 +18,20 @@ type OpcUAWorkarounds struct {
|
||||||
AdditionalValidStatusCodes []string `toml:"additional_valid_status_codes"`
|
AdditionalValidStatusCodes []string `toml:"additional_valid_status_codes"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type ConnectionState opcua.ConnState
|
||||||
|
|
||||||
|
const (
|
||||||
|
Closed ConnectionState = ConnectionState(opcua.Closed)
|
||||||
|
Connected ConnectionState = ConnectionState(opcua.Connected)
|
||||||
|
Connecting ConnectionState = ConnectionState(opcua.Connecting)
|
||||||
|
Disconnected ConnectionState = ConnectionState(opcua.Disconnected)
|
||||||
|
Reconnecting ConnectionState = ConnectionState(opcua.Reconnecting)
|
||||||
|
)
|
||||||
|
|
||||||
|
func (c ConnectionState) String() string {
|
||||||
|
return opcua.ConnState(c).String()
|
||||||
|
}
|
||||||
|
|
||||||
type OpcUAClientConfig struct {
|
type OpcUAClientConfig struct {
|
||||||
Endpoint string `toml:"endpoint"`
|
Endpoint string `toml:"endpoint"`
|
||||||
SecurityPolicy string `toml:"security_policy"`
|
SecurityPolicy string `toml:"security_policy"`
|
||||||
|
|
@ -73,29 +87,15 @@ func (o *OpcUAClientConfig) CreateClient(log telegraf.Logger) (*OpcUAClient, err
|
||||||
Log: log,
|
Log: log,
|
||||||
}
|
}
|
||||||
c.Log.Debug("Initialising OpcUAClient")
|
c.Log.Debug("Initialising OpcUAClient")
|
||||||
c.State = Disconnected
|
|
||||||
|
|
||||||
err = c.setupWorkarounds()
|
err = c.setupWorkarounds()
|
||||||
return c, err
|
return c, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// ConnectionState used for constants
|
|
||||||
type ConnectionState int
|
|
||||||
|
|
||||||
const (
|
|
||||||
// Disconnected constant State 0
|
|
||||||
Disconnected ConnectionState = iota
|
|
||||||
// Connecting constant State 1
|
|
||||||
Connecting
|
|
||||||
// Connected constant State 2
|
|
||||||
Connected
|
|
||||||
)
|
|
||||||
|
|
||||||
type OpcUAClient struct {
|
type OpcUAClient struct {
|
||||||
Config *OpcUAClientConfig
|
Config *OpcUAClientConfig
|
||||||
Log telegraf.Logger
|
Log telegraf.Logger
|
||||||
|
|
||||||
State ConnectionState
|
|
||||||
Client *opcua.Client
|
Client *opcua.Client
|
||||||
|
|
||||||
opts []opcua.Option
|
opts []opcua.Option
|
||||||
|
|
@ -164,10 +164,7 @@ func (o *OpcUAClient) Connect() error {
|
||||||
|
|
||||||
switch u.Scheme {
|
switch u.Scheme {
|
||||||
case "opc.tcp":
|
case "opc.tcp":
|
||||||
o.State = Connecting
|
if err := o.SetupOptions(); err != nil {
|
||||||
|
|
||||||
err = o.SetupOptions()
|
|
||||||
if err != nil {
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -184,11 +181,8 @@ func (o *OpcUAClient) Connect() error {
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(o.Config.ConnectTimeout))
|
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(o.Config.ConnectTimeout))
|
||||||
defer cancel()
|
defer cancel()
|
||||||
if err := o.Client.Connect(ctx); err != nil {
|
if err := o.Client.Connect(ctx); err != nil {
|
||||||
o.State = Disconnected
|
|
||||||
return fmt.Errorf("error in Client Connection: %w", err)
|
return fmt.Errorf("error in Client Connection: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
o.State = Connected
|
|
||||||
o.Log.Debug("Connected to OPC UA Server")
|
o.Log.Debug("Connected to OPC UA Server")
|
||||||
|
|
||||||
default:
|
default:
|
||||||
|
|
@ -206,7 +200,6 @@ func (o *OpcUAClient) Disconnect(ctx context.Context) error {
|
||||||
|
|
||||||
switch u.Scheme {
|
switch u.Scheme {
|
||||||
case "opc.tcp":
|
case "opc.tcp":
|
||||||
o.State = Disconnected
|
|
||||||
// We can't do anything about failing to close a connection
|
// We can't do anything about failing to close a connection
|
||||||
err := o.Client.CloseWithContext(ctx)
|
err := o.Client.CloseWithContext(ctx)
|
||||||
o.Client = nil
|
o.Client = nil
|
||||||
|
|
@ -215,3 +208,10 @@ func (o *OpcUAClient) Disconnect(ctx context.Context) error {
|
||||||
return fmt.Errorf("invalid controller")
|
return fmt.Errorf("invalid controller")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (o *OpcUAClient) State() ConnectionState {
|
||||||
|
if o.Client == nil {
|
||||||
|
return Disconnected
|
||||||
|
}
|
||||||
|
return ConnectionState(o.Client.State())
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -90,29 +90,26 @@ func (o *ReadClient) Connect() error {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (o *ReadClient) ensureConnected() error {
|
func (o *ReadClient) ensureConnected() error {
|
||||||
if o.State == opcua.Disconnected {
|
if o.State() == opcua.Disconnected {
|
||||||
err := o.Connect()
|
return o.Connect()
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (o *ReadClient) CurrentValues() ([]telegraf.Metric, error) {
|
func (o *ReadClient) CurrentValues() ([]telegraf.Metric, error) {
|
||||||
err := o.ensureConnected()
|
if err := o.ensureConnected(); err != nil {
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
err = o.read()
|
if state := o.State(); state != opcua.Connected {
|
||||||
if err != nil && o.State == opcua.Connected {
|
return nil, fmt.Errorf("not connected, in state %q", state)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := o.read(); err != nil {
|
||||||
// We do not return the disconnect error, as this would mask the
|
// We do not return the disconnect error, as this would mask the
|
||||||
// original problem, but we do log it
|
// original problem, but we do log it
|
||||||
disconnectErr := o.Disconnect(context.Background())
|
if derr := o.Disconnect(context.Background()); derr != nil {
|
||||||
if disconnectErr != nil {
|
o.Log.Debug("Error while disconnecting: ", derr)
|
||||||
o.Log.Debug("Error while disconnecting: ", disconnectErr)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue