diff --git a/examples/test-adctk_hello_world.sh b/examples/test-adctk_hello_world.sh new file mode 100755 index 0000000..b38a437 --- /dev/null +++ b/examples/test-adctk_hello_world.sh @@ -0,0 +1,28 @@ +#! /bin/bash +module purge +module load aue/gdb +rm -rf test.outputs adc.file_plugin.log +mkdir test.outputs +. ./test-env-publishers +export ADC_MULTI_PUBLISHER_DEBUG=1 +export ADC_MULTIFILE_PLUGIN_DEBUG=1 +env |grep ADC |grep CURL +env |grep ADC |grep MULTI_PUB +echo starting... +adctk_hello_world > >(tee test.outputs/hello-world.stdout.log) 2> >(tee test.outputs/hello-world.stderr.log) +if test -s ./adc.file_plugin.log; then + echo "got ./adc.file_plugin.log" +else + echo "no ./adc.file_plugin.log" +fi +if test -s test.outputs/out.file.log; then + echo got test.outputs/out.file.log +else + echo "no test.outputs/out.file.log" +fi +sleep 1 ; # catch up with nfs delays +if test -s ./test.outputs/script; then + echo "got test.outputs/script" +else + echo "no test.outputs/script" +fi diff --git a/examples/test-env-publishers b/examples/test-env-publishers index 3d0d5ca..f4a486e 100644 --- a/examples/test-env-publishers +++ b/examples/test-env-publishers @@ -11,11 +11,12 @@ TEST_LDMS_AUTH=munge TEST_LDMS_PORT=412 # define the plugins for the multipublisher -TEST_ALL_PUBS="curl:file:multifile:script:stdout:syslog:ldmsd_stream_publish:ldms_message_publish" +TEST_ALL_PUBS="http:file:multifile:script:stdout:syslog:ldmsd_stream_publish:ldms_message_publish" TEST_LOCAL_PUBS="file:script:stdout:multifile" -TEST_NET_PUBS="curl:syslog:ldmsd_stream_publish:ldms_message_publish" +TEST_NET_PUBS="http:syslog:ldmsd-system-message:ldms-system-stream" #serial -TEST_ASF_PUBS="multifile:ldms_message_publish" +TEST_ASF_PUBS="multifile:ldms-system-message:http" +TEST_ASF_PUBS="ldms-system-message" # default the multi_publisher to node-local ones; can be overridden. export ADC_MULTI_PUBLISHER_NAMES=$TEST_ASF_PUBS @@ -48,7 +49,7 @@ export ADC_CURL_PLUGIN_DIRECTORY="/dev/shm/adc" export ADC_CURL_PLUGIN_PROG="/usr/bin/curl" export ADC_CURL_PLUGIN_PORT="443" export ADC_CURL_PLUGIN_URL="$TEST_ADC_URL" -export ADC_CURL_PLUGIN_DEBUG="0" +export ADC_CURL_PLUGIN_DEBUG="2" # environment variables for file serial code plugin export ADC_FILE_PLUGIN_DIRECTORY="." @@ -58,12 +59,12 @@ export ADC_FILE_PLUGIN_DEBUG="0" # environment variables for file parallel code plugin export ADC_MULTIFILE_PLUGIN_DIRECTORY=$ADC_FS_ROOT/$WGTEAM -export ADC_MULTIFILE_PLUGIN_DEBUG="1" export ADC_MULTIFILE_PLUGIN_DEBUG="0" +export ADC_MULTIFILE_PLUGIN_DEBUG="1" # Normally rank should be defined in code rather than the environment, # using either setenv before creating the plugin instance # or passing rank to the publisher config setup. -export ADC_MULTIFILE_PLUGIN_RANK="" +export ADC_MULTIFILE_PLUGIN_RANK="0" # environment variables for ldmsd_stream_publish subprocess plugin export ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_DIRECTORY="/dev/shm/adc" @@ -73,7 +74,7 @@ export ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_AUTH="$TEST_LDMS_AUTH" export ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_PORT="$TEST_LDMS_PORT" export ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_HOST="localhost" export ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_AFFINITY="all" -export ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_DEBUG="0" +export ADC_LDMS_MESSAGE_PUBLISH_PLUGIN_DEBUG="1" # environment variables for ldmsd_stream_publish subprocess plugin export ADC_LDMSD_STREAM_PUBLISH_PLUGIN_DIRECTORY="/dev/shm/adc" diff --git a/examples/test-test_adctk_builder.sh b/examples/test-test_adctk_builder.sh new file mode 100755 index 0000000..4137957 --- /dev/null +++ b/examples/test-test_adctk_builder.sh @@ -0,0 +1,28 @@ +#! /bin/bash +module purge +module load aue/gdb +rm -rf test.outputs adc.file_plugin.log +mkdir test.outputs +. ./test-env-publishers +export ADC_MULTI_PUBLISHER_DEBUG=1 +export ADC_MULTIFILE_PLUGIN_DEBUG=1 +env |grep ADC +echo starting... +test_adctk_builder > >(tee test.outputs/stdout.log) 2> >(tee test.outputs/stderr.log) +#test_adctk_builder |tee test.outputs/test_adctk_builder.out +if test -s ./adc.file_plugin.log; then + echo "got ./adc.file_plugin.log" +else + echo "no ./adc.file_plugin.log" +fi +if test -s test.outputs/out.file.log; then + echo got test.outputs/out.file.log +else + echo "no test.outputs/out.file.log" +fi +sleep 1 ; # catch up with nfs delays +if test -s ./test.outputs/script; then + echo "got test.outputs/script" +else + echo "no test.outputs/script" +fi diff --git a/pyproject.toml b/pyproject.toml index 51163d0..b644bc8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,7 @@ dev = [ [project.scripts] adctk-publish = "adctk.scripts.adctk_publish:main" +adctk_hello_world = "adctk.scripts.adctk_hello_world:main" test_adc_types = "adctk.scripts.test_adc_types:main" test_adctk_builder = "adctk.scripts.test_adctk_builder:main" test_old_builder = "adctk.scripts.test_builder:main" diff --git a/src/adctk/builder.py b/src/adctk/builder.py index df363c1..1e51d6f 100644 --- a/src/adctk/builder.py +++ b/src/adctk/builder.py @@ -879,7 +879,7 @@ def add(self, name: str, value: any, value_type=None): if name is None or value is None: raise ValueError("add: name and value cannot be None") t = type(value) - print(f"\nadd {name} {t}") + ### print(f"\n DEB: add {name} {t}") if isinstance(t, Builder): self.add_section(name, value) return diff --git a/src/adctk/factory.py b/src/adctk/factory.py index 0a0f1b6..9460ef7 100644 --- a/src/adctk/factory.py +++ b/src/adctk/factory.py @@ -128,10 +128,7 @@ def get_multi_publisher(strict=False, "http":{"port": "8443", "url": "localhost"}}) """ -# try: return publisher.MultiPublisher(strict, plugins) -# except: -# return None def __init__(self): pass # Factory.__init__ nothing to do diff --git a/src/adctk/publisher.py b/src/adctk/publisher.py index 33e1080..736fcdb 100644 --- a/src/adctk/publisher.py +++ b/src/adctk/publisher.py @@ -261,7 +261,7 @@ def _config_from_list(self, plugins: list[str], strict: bool): def _config_from_env(self, env_name: str, strict: bool): """@brief configure with list following env_name.split(":")""" env_val = os.getenv(env_name, "") - if env_val is None or len(env_val): + if env_val is None or not len(env_val): return plugins = env_val.split(":") self._config_from_list(plugins, strict) @@ -303,6 +303,7 @@ def add(self, publisher: Publisher) -> None: if not publisher: return None self.pvec.append(publisher) + print(f"DEB: Adding publisher {publisher.name()}") return None def publish(self, b: builder.Builder) -> int: diff --git a/src/adctk/scripts/adctk_hello_world.py b/src/adctk/scripts/adctk_hello_world.py new file mode 100644 index 0000000..45bc9df --- /dev/null +++ b/src/adctk/scripts/adctk_hello_world.py @@ -0,0 +1,95 @@ +# Copyright 2026 NTESS. See the top-level LICENSE.txt file for details. +# +# SPDX-License-Identifier: BSD-3-Clause + +import os +import adctk +#*! \file adcHelloWorld.py +# This demonstrates using the adctk.factory API to build and publish a message. +# The message sent includes the bare minimum, plus hello world and host data. + +#! \addtogroup examples +# @{ + +#! +# \brief adctk python hello world without hard-coded publisher choices. +def main() -> int: + """! + @brief adctk python hello world without hard-coded publisher choices. + """ + + # create a factory + f = adctk.Factory() + + # create a message and add header + b = f.get_builder() + b.add_header_section("cxx_demo_1") + + # add an application-defined payload to the message + app_data = f.get_builder() + app_data.add("hello", "world") + b.add_app_data_section(app_data) + + # add environment chunks of interest on at least the first message in production + b.add_host_section(all_hs=True) + + # could add lots of other sections, as needed. + + print(f'adc pub version: {adctk.Publisher.API_VERSION["version"]}' ) + print(f'adc builder version: {adctk.Builder.API_VERSION["version"]}' ) + + # create publishers following runtime environment variables and defaults. + # Do not tolerate failures in a testing environment. + mp = f.get_multi_publisher(strict=True, plugins=[]) + + # send built message b to all publishers + err = mp.publish(b) + if err: + print(f"got {err} publication errors." ) + + # do some work + + # send the final status updating the header to get the needed new timestamp and uuid + b.add_exit_data_section(0, "all good in python", None) + b.add_header_section("cxx_demo_1") + err = mp.publish(b) + if err: + print(f"got {err} publication errors." ) + + # clean up all publishers + mp.terminate() + + # the next block is skipped pending implementation of get_multifile_log_path + # the next block is skipped pending implementation of consolidate_multifile_logs + if False: + # may need to sleep here to give local fs a chance to catch up + # dir/user/[wfid.].host.Ppid.Tstarttime.pptr/application.Rrank.XXXXXX + # --> + # dir/user/consolidated.[wfid].adct-json.multi.xml + path = os.getenv("ADC_MULTIFILE_PLUGIN_DIRECTORY") + wfid = os.getenv("ADC_WFID") + ## pattern = adctk.get_multifile_log_path(path, wfid) + + old_paths = [] + ## new_files = adctk.consolidate_multifile_logs(pattern, old_paths) + new_files = [] + if len(old_paths): + for i in old_paths: + print(f"consolidating from: {i}") + if False: + try: + os.remove(i) # we could delete the merged files. + except FileNotFoundError: + pass + for i in new_files: + print(f"consolidated to: {i}" ) + else: + print("no consolidation done.") + + return 0 + +#! @} +#! @} + +if __name__ == "__main__": + main() diff --git a/src/adctk/scripts/test_adc_types.py b/src/adctk/scripts/test_adc_types.py index 8a0be91..4e0a87d 100644 --- a/src/adctk/scripts/test_adc_types.py +++ b/src/adctk/scripts/test_adc_types.py @@ -16,7 +16,7 @@ def add2(x, name, value): if isinstance(value, range): pass if type(value) == int: - if fits_int64(value): + if adctk.adc_types.fits_int64(value): pass #store as cp_number_str pass diff --git a/src/adctk/scripts/test_adctk_builder.py b/src/adctk/scripts/test_adctk_builder.py index 2501da9..14dd89f 100644 --- a/src/adctk/scripts/test_adctk_builder.py +++ b/src/adctk/scripts/test_adctk_builder.py @@ -2,8 +2,8 @@ # # SPDX-License-Identifier: BSD-3-Clause # -import numpy import array +import numpy import adctk #import adctk.adc_types #import adctk.builder @@ -24,23 +24,23 @@ # config w/file_config and call initialize. def test_publisher( pub: adctk.Publisher, b: adctk.Builder) -> int : - err = 0; - e = 0; + err = 0 + e = 0 err = pub.config(**file_config) if err: print(f"config failed {err}") - e += 1; + e += 1 err = pub.initialize() if err: print(f"initialize failed {err}") - e += 1; - err = 0; - err = pub.publish(b); # there should be 1 b in the output + e += 1 + err = 0 + err = pub.publish(b) # there should be 1 b in the output if err: print(f"publish 1 failed {err}") - e += 1; - err = 0; - return e; + e += 1 + err = 0 + return e def not_scalar_or_string(var): # returns False for numeric scalars and standard strings @@ -153,7 +153,7 @@ def roundtrip_array(name: str, val: any, cptype: adctk.adc_types.ScalarType, cou print(f"[{i}]: {ck.data[i]} vs {val[i]}") if not bad: print(" ok") - return; + return print("\nBAD", end='') print(f"\tkt: {ck.kt}", end='') print(f"\tst: {str(ck.st)}", end='') @@ -207,7 +207,7 @@ def populate_builder(b, f): u8 = numpy.iinfo(numpy.int8).max u16 = numpy.iinfo(numpy.int16).max u32 = numpy.iinfo(numpy.int32).max - u64 = numpy.iinfo(numpy.uint64).max; + u64 = numpy.iinfo(numpy.uint64).max b.add("u8", u8, 'uint8') roundtrip("u8", u8, adctk.adc_types.ScalarType.cp_uint8, b) b.add("u16", u16, 'uint16') @@ -261,23 +261,23 @@ def populate_builder(b, f): b.add("cppstr", cppstr) roundtrip_string("cppstr", cppstr, adctk.adc_types.ScalarType.cp_cstr, b) - cstr = "cstr_nul"; + cstr = "cstr_nul" b.add("cstr1", cstr) roundtrip_string("cstr1", cstr, adctk.adc_types.ScalarType.cp_cstr, b) - jstr = "{\"a\":\"b\", \"c\":[1,2, 3]}"; + jstr = "{\"a\":\"b\", \"c\":[1,2, 3]}" b.add_json_string("jstr1", jstr) roundtrip_string("jstr1", jstr, adctk.adc_types.ScalarType.cp_json_str, b) - ystr = "---\na: b\nc: [1,2, 3]\nd:\n e: 1\n f: 2"; + ystr = "---\na: b\nc: [1,2, 3]\nd:\n e: 1\n f: 2" b.add_yaml_string("ystr1", ystr) roundtrip_string("ystr1", ystr, adctk.adc_types.ScalarType.cp_yaml_str, b) - xstr = " Tove Jani "; + xstr = " Tove Jani " b.add_xml_string("xstr1", xstr) roundtrip_string("xstr1", xstr, adctk.adc_types.ScalarType.cp_xml_str, b) - nstr = "1234567890123456789012345678901234567890.123"; + nstr = "1234567890123456789012345678901234567890.123" b.add_number_string("number1", nstr) roundtrip_string("number1", nstr, adctk.adc_types.ScalarType.cp_number_str, b) ss = b.serialize() @@ -378,13 +378,13 @@ def populate_builder(b, f): cpu2 = dict() mem2 = dict() - host2["NAME"] = "MYHOST"; - host2["CLUSTER"] ="MYCLUSTER"; - cpu2["PROCESSOR"] = "PENTIUM ii"; - mem2["SIZE"] = "256g"; - arch2["CPU"] = cpu2; - arch2["MEMORY"] = mem2; - host2["ARCHITECTURE"] = arch2; + host2["NAME"] = "MYHOST" + host2["CLUSTER"] ="MYCLUSTER" + cpu2["PROCESSOR"] = "PENTIUM ii" + mem2["SIZE"] = "256g" + arch2["CPU"] = cpu2 + arch2["MEMORY"] = mem2 + host2["ARCHITECTURE"] = arch2 b.add("host2", host2) @@ -434,7 +434,7 @@ def main() -> int: mp.publish(b) mp.terminate() - return 0; + return 0 if __name__ == "__main__": main()