Client
Description
Implements the client methods that a process needs in order to to interact with the message bus.
Synopsis
use Assure1::MessageBus::Client;
my $MBClient = Assure1::MessageBus::Client($baseDir, \[{OPTIONS}\]);
my $response = $mbClient->ClientRequest(METHOD, \[ARGUMENTS\]);
Constructor
new
Create new client for sending and receiving messages
new(\%options)
Options
log -> Assure1::Log instance, omitt to disable logging
config -> Assure1::Config instance
workqueue -> Optional reference to a Thread::Queue to send incoming messages to for processing, if not given will ignore any incoming broker messages that are not replies.
Incoming messages are queued in the following frozen format (use Storable):
body - message body; this can contain any number of formats such as a hash, array, or string
type - message type (either "direct" or "failover")
rawmessage - raw message info including message properties; use when sending replies through reply() as 'orig_msg'
bindings -> Additional exchanges to bind to (e.g. EX_THRESHOLD)
timeout -> Time to wait for response before timing out. Default 30 seconds
server -> Specific broker to talk to; should be the FQDN. Defaults to to local hostfqdn
blocking -> set whether to block on ClientRequest and GetResponse. Defaults to 1
Returns
Assure1::MessageBus::Client object
Synopsis
my $MBQueue = new Thread::Queue;
my $MBClient = Assure1::MessageBus::Client->new({
log => $Log,
config => $config,
workqueue => \$MBQueue
});
threads->new(sub {
# if we can't start the handler thread we want to terminate the app
threads->set_thread_exit_only(0);
eval {
local $SIG{__DIE__};
$MBClient->handle();
};
if ($@) {
$Self->{Log}->Message('FATAL', '{SIGDIE}: ' . $@);
}
})->detach();
Methods
ClientRequest
Generic client request to call an RPC method with arguments.
ClientRequest(\%options)
Options
method - RPC method to call
arguments - optional arguments to pass with the request
server - optional recipient server, defaults to the value of server() which defaults to local Unified Assurance broker
blocking - optional returns after sending request without waiting for response. Defaults to current blocking() setting. Response can be retrieved later with a GetResponse() if necessary.
Returns
Hashref containing the following response keys:
Success -> 1 for Success, 0 for Failure, or -1 for in non-blocking mode and no response= yet
Message -> RPC response message, error text if failed
Data -> Data returned or empty array if nothing. If blocking = 1 returns the message ID with which to use with a subsequent GetResponse() call
GetResponse
GetResponse to a given original message.
GetResponse($MessageID, $Blocking)
Arguments
MessageID - MessageID to check response for
Blocking - optional: 1 to wait for response, 0 to return data if ready or without waiting, defaults to current blocking() setting
Returns
Hashref containing the following response keys:
Success -> 1 for Success, 0 for Failure, or -1 when in non-blocking mode and no response yet
Message -> RPC response data, error text if failed, or if blocking = 1 returns the message ID
Synopsis
# Sending a message to local broker
Broker::Client->Reply({
orig_msg => $msg,
reply => {success => 1, message => 'returning 2 items', data => [1,2]}
});
Reply
Reply to a given original message.
Reply(\%options)
Options
reply - Reply hash, gets converted to a JSON string. must contain the following keys:
success - boolean success or failure
message - optional message, if success=false, this should be the error message and data should be blank
data - array of data to return if any; items can be any data type except blessed objects, even hashes or arrays.
orig_msg - Original message information the reply is for
Synopsis
# Sending a message to local broker
Broker::Client->Reply({
orig_msg => $msg,
reply => {success => 1, message => 'returning 2 items', data => [1,2]}
});
SendHealthMetric
Send a health metric to InfluxDB via Telegraf
SendHealthMetric(\%options)
Options
measurement -> Metric Type Name
tags -> hash reference of:
host -> (optional) Device Host (defaults to local HostFQDN)
instanceName -> Instance Name (e.g. Application Name)
fields -> hash reference of:
value -> value to send
utilization -> (optional) Utilization of value if a max value is known
epochTime -> (optional) Epoch time collected (defaults to time()). Strongly recommend using the epoch from time() rounded to nearest poll time
SendThresholdViolation
Queue a threshold violation.
SendThresholdViolation(\%options)
Options
ThresholdType -> PollerThresholds.ThresholdTypeID (1 => Standard, 10 => Trend % Change, 14 => Abnormal, 16 => Missing Data, 20 => Trend Prediction)
ThresholdName -> PollerThresholds.ThresholdName
ThresholdMessage -> PollerThresholds.Message
ThresholdTimeRange -> PollerThresholds.TimeRange
ThresholdOperator -> PollerThresholds.Warning|CriticalOperator
ThresholdValue -> PollerThresholds.Warning|CriticalValue
ThresholdSeverity -> PollerThresholds.Warning|CriticalSeverity
Time -> EpochTimeStamp
Measurement -> PollerThresholds.Measurement
MetricValue -> InfluxDB measurement Mean value|availability|utilization (from MetricField)
DeviceName -> InfluxDB measurement deviceName
InstanceName -> InfluxDB measurement instance
DeviceZoneID -> InfluxDB measurement devicezone
blocking
Get current block status or set whether to block waiting for responses.
blocking([$block])
clusterID
Get current clusterID or set clusterID for clustered polling/load balancing. Should be the FailoverName given when joining a pool
clusterID([$clusterID])
close
Close the current connection to the message bus
close()
server
Get current server or set server to send any ClientRequests to. Should be the Unified Assurance broker FQDN.
server([$server])
workqueue
Get current workqueue or set workqueue. Should be a reference to a Thread::Queue
workqueue([$workqueue])