adctk.ldms_publisher.LDMSPipeMessagePublisher Class Reference

Publish data over a pipe to ldms_message_publish once via pipe recreated at each call. More...

Inheritance diagram for adctk.ldms_publisher.LDMSPipeMessagePublisher:

Collaboration diagram for adctk.ldms_publisher.LDMSPipeMessagePublisher:

Public Member Functions
	__init__ (self)
int	config (self, **kwargs)
	Configure the plugin with the options given.
int	initialize (self)
	Set up any resources needed for reuse in publish().
	publish (self, builder.Builder b)
	finalize (self)
	Clean up any resources set up in initialize (files, pipes, etc).
Public Member Functions inherited from adctk.publisher.Publisher
str	name (self)
	Get the name which a factory or error message should use.
str	version (self)
dict[str, str]	get_option_defaults (self)
None	pause (self)
	Cause publish calls to be ignored until resume is called.
None	resume (self)
	Cancel an prior calls to pause.
int	publish_str (self, str json_string)
int	publish_obj (self, obj)

Public Attributes
	plugin_prefix
	defaults
	prog
	port
	auth
	tag
	host
	timeout
	mode
	state
Public Attributes inherited from adctk.publisher.Publisher
	defaults
	plugin_prefix
	paused
	state
	mode

Protected Member Functions
int	_post (self, msg)
Protected Member Functions inherited from adctk.publisher.Publisher
str	_get (self, dict[str, str] opts, str field)
bool	_get_bool (self, dict[str, str] opts, str field)

Static Protected Attributes
str	_name = "ldms-pipe-message"
str	_version = "1.0.0"
Static Protected Attributes inherited from adctk.publisher.Publisher
str	_name = "unknown"
str	_version = "0.0.0"

Additional Inherited Members
Static Public Member Functions inherited from adctk.publisher.Publisher
tuple[float]	get_timeout_defaults ()
	Get the adc server connect and response timeouts defined in the environment ADC_SERVER_CONNECT_TIMEOUT and ADC_SERVER_POST_TIMEOUT as a tuple or default values if not defined.
str	get_url_default ()
	Get the alleged adc server url base from environment ADC_SERVER_URL or the default if it is not present.
str	get_port_default ()
	Get the adc server port number to use if it is properly defined in the environment ADC_SERVER_PORT or the default if it is not.
Static Public Attributes inherited from adctk.publisher.Publisher
	API_VERSION = adc_version.AdcVersion()._dict()

Detailed Description

Publish data over a pipe to ldms_message_publish once via pipe recreated at each call.

Environment Variables

• ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_PROG
• ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_TAG
• ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_AUTH
• ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_PORT
• ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_HOST
• ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_TIMEOUT

Constructor & Destructor Documentation

__init__()

adctk.ldms_publisher.LDMSPipeMessagePublisher.__init__

(

self

)

Reimplemented from adctk.publisher.Publisher.

Reimplemented in adctk.ldms_publisher.LDMSSystemMessagePublisher.

    def __init__(self):
        super().__init__()
        self.plugin_prefix = "ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_"
        self.defaults = _defaults_pipe_message
        self.prog: str = None
        self.port: str = None
        self.auth: str = None
        self.timeout: int = 2
        self.tag: str = None
        self.host: str = None
 

Member Function Documentation

_post()

int adctk.ldms_publisher.LDMSPipeMessagePublisher._post (   self,   msg  )

protected

Reimplemented in adctk.ldms_publisher.LDMSSystemMessagePublisher.

    def _post(self, msg) ->int:
        try:
            cmd = [self.prog,
                   "-t", "json",
                   "-x", "sock",
                   "-p", self.port,
                   "-a", self.auth,
                   "-m", self.tag,
                   "-h", self.host
                   ]
            log.logger.info("LDMSPipe-publisher command: %s", cmd)
            process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                        text=True)
            process.stdin.write(msg)
            process.stdin.flush()
            process.stdin.close()
            stdout, stderr = process.communicate(timeout=self.timeout)
            output_string = stdout.decode("utf-8")
            error_string = stderr.decode("utf-8")
            log.logger.debug("Output: %s Error: %s", output_string, error_string)
            ret = process.returncode
            if ret != 0:
                log.logger.warning("LDMSPipe-publisher: publish process failed with error: %d",
                                    ret)
        except subprocess.TimeoutExpired:
            process.kill()
            stdout, stderr = process.communicate()
            log.logger.warning("LDMSPipe-publisher: timeout expired")
        return ret
 

config()

int adctk.ldms_publisher.LDMSPipeMessagePublisher.config	(		self,
		**	kwargs
	)

Configure the plugin with the options given.

For plugin QQQ, Environment variables ADC_QQQ_PLUGIN_<key> will override the plugin default for any key not defined in kwargs. Here QQQ is plugin.name().upper(). Generic options:

Parameters

kwargs	If env_prefix is included in kwargs, replaces plugin environment variables with those which are prefixed using env_prefix.
env_prefix	Typically, env_prefix is set to "PPP_ADC_QQQ_PLUGIN_" if application PPP wants to override the defaults of adctk plugin QQQ. Here QQQ is plugin.name().upper(). If no prefixed version of the variable is present, the plugin default is used. kwargs takes precedence over environment variables. Plugin defaults are used if not overridden by kwargs or an environment variable.

Returns

int non-zero if configuration cannot be completed using the arguments, environment, and default.

Reimplemented from adctk.publisher.Publisher.

Reimplemented in adctk.ldms_publisher.LDMSSystemMessagePublisher.

    def config(self, **kwargs) -> int:
        self.prog = self._get(kwargs, "PROG")
        self.port = self._get(kwargs, "PORT")
        self.auth = self._get(kwargs, "AUTH")
        self.tag = self._get(kwargs, "TAG")
        self.host = self._get(kwargs, "HOST")
        timeout = self._get(kwargs, "TIMEOUT")
        try:
            x = int(timeout)
            self.timeout = x
        except ValueError:
            log.logger.error("LDMSPipe-publisher TIMEOUT %s is invalid. integer needed.", timeout)
        return 0
 

finalize()

adctk.ldms_publisher.LDMSPipeMessagePublisher.finalize

(

self

)

Clean up any resources set up in initialize (files, pipes, etc).

Reimplemented from adctk.publisher.Publisher.

    def finalize(self):
        pass # nothing to do here. LDMSPipeMessagePublisher.finalize
 

initialize()

int adctk.ldms_publisher.LDMSPipeMessagePublisher.initialize

(

self

)

Set up any resources needed for reuse in publish().

Reimplemented from adctk.publisher.Publisher.

    def initialize(self) -> int:
        if self.mode == publisher.Publisher.Mode.CONFIG:
            self.config()
        if self.mode != publisher.Publisher.Mode.INIT:
            self.state = publisher.Publisher.State.ERR
            return errno.EINVAL
        self.mode = publisher.Publisher.Mode.PUB_OR_FINAL
        return 0
 

publish()

adctk.ldms_publisher.LDMSPipeMessagePublisher.publish	(		self,
		builder.Builder	b
	)

publish json extracted from builder to _base_url server

Reimplemented from adctk.publisher.Publisher.

    def publish(self, b: builder.Builder):
        """publish json extracted from builder to _base_url server"""
        s = b.get_json_dump()
        ret = self._post(s)
        return ret
 

Member Data Documentation

_name

str adctk.ldms_publisher.LDMSPipeMessagePublisher._name = "ldms-pipe-message"

staticprotected

_version

str adctk.ldms_publisher.LDMSPipeMessagePublisher._version = "1.0.0"

staticprotected

auth

adctk.ldms_publisher.LDMSPipeMessagePublisher.auth

defaults

adctk.ldms_publisher.LDMSPipeMessagePublisher.defaults

host

adctk.ldms_publisher.LDMSPipeMessagePublisher.host

mode

adctk.ldms_publisher.LDMSPipeMessagePublisher.mode

plugin_prefix

adctk.ldms_publisher.LDMSPipeMessagePublisher.plugin_prefix

port

adctk.ldms_publisher.LDMSPipeMessagePublisher.port

prog

adctk.ldms_publisher.LDMSPipeMessagePublisher.prog

state

adctk.ldms_publisher.LDMSPipeMessagePublisher.state

tag

adctk.ldms_publisher.LDMSPipeMessagePublisher.tag

timeout

adctk.ldms_publisher.LDMSPipeMessagePublisher.timeout

---

The documentation for this class was generated from the following file:

• src/adctk/ldms_publisher.py