ldms_blob_file_to_http_post Namespace Reference

Classes
class	Blob
class	BlobReader
class	BlobScanner
class	PostFilter
class	Readline0

Functions
	readline0 (typing.Union[typing.BinaryIO, int] file_=sys.stdin.buffer, bytes separator=b'\0', int blocksize=2 **16)
	process_file (fn, url, completed_dir, retry_dir, error_dir, log_dir, retry, validate)
	handle_request_status (returned_data)
	main ()

Function Documentation

handle_request_status()

ldms_blob_file_to_http_post.handle_request_status

(

returned_data

)

def handle_request_status(returned_data):
    status_code = returned_data.status_code
    if status_code < 200 or status_code > 300:
          logger.error(f"Status Code: {status_code}, Message: {returned_data.raw}")
          raise HttpStatusCodeError(status_code)
    logger.info(f"Posting successful with status Code: {status_code}")
 
 

main()

ldms_blob_file_to_http_post.main

(

)

def main():
    parser = argparse.ArgumentParser(description="Generate ADC web posts from ldms json stream files.")
    parser.add_argument("--url", default=None, help="where to post messages (or full path for debug output)")
    try:
        parser.add_argument("--validate", default=False, help="verify input is well-formed json before posting", action=argparse.BooleanOptionalAction)
        parser.add_argument("--retry", default=False, help="whether to publish the retry indices found in retry-dir files instead of all indices", action=argparse.BooleanOptionalAction)
    except:
        print("Need python with argparse.BooleanOptionalAction; usually py >= 3.9")
        return 1
    parser.add_argument("--completed-dir", default=None, help="where to put index numbers of messages that the server accepted that were not json.")
    parser.add_argument("--retry-dir", default=None, help="where to put index numbers and files of json messages the server was unavailable for or unhappy with.")
    parser.add_argument("--error-dir", default=None, help="where to log index numbers of json messages that provoked errors")
    parser.add_argument("--log-dir", default=None, help="log file directory")
    parser.add_argument('files', nargs='+')
    args = parser.parse_args()
    if args.retry:
        print("ERROR: retry not implemented yet")
        sys.exit(1)
    if args.completed_dir and not os.path.isdir(args.completed_dir):
        print("ERROR: completed-dir %s missing" % args.completed_dir)
        sys.exit(1)
    if args.retry_dir and not os.path.isdir(args.retry_dir):
        print("ERROR: retry-dir %s is missing" % args.retry_dir)
        sys.exit(1)
    if args.error_dir and not os.path.isdir(args.error_dir):
        print("ERROR: error-dir %s is missing" % args.error_dir)
        sys.exit(1)
    if args.log_dir and not os.path.isdir(args.log_dir):
        print("ERROR: log-dir %s is missing" % args.log_dir)
        sys.exit(1)
    for f in args.files:
        if args.validate:
            print("with validation");
        print("Processing " + f)
        if not args.completed_dir:
            compdir = os.path.join(os.path.dirname(f),"COMPLETED")
        else:
            compdir = args.completed_dir
        if not args.retry_dir:
            retdir = os.path.join(os.path.dirname(f),"RETRY")
        else:
            retdir = args.retry_dir
        # Given the output naming conventions, retdir, compdir can be the same, though this might
        # make future glob processing really slow.
        # files cannot be processed in place unless it is a retry run (which is not yet implemented)
        maxerr = process_file(f, args.url, compdir, retdir, args.error_dir, args.log_dir, args.retry, args.validate)
        if maxerr < 2:
            print("Processed {}".format(f))
        else:
            print("Retry maybe needed for {} in {}: maxerr {}".format(os.path.basename(f), retdir, maxerr))
 
    return 0
 

process_file()

ldms_blob_file_to_http_post.process_file	(		fn,
			url,
			completed_dir,
			retry_dir,
			error_dir,
			log_dir,
			retry,
			validate
	)

 @return maximum error code seen in add function calls. if > 1,
file retry later may be useful.

def process_file(fn, url, completed_dir, retry_dir, error_dir, log_dir, retry, validate):
    """ @return maximum error code seen in add function calls. if > 1,
    file retry later may be useful."""
    br = BlobReader(fn)
    pf = PostFilter(br, url, completed_dir, retry_dir, error_dir, log_dir, validate)
    if br.timing:
        print("total timing: %d " % (len(br.timing)//2))
        # print("%g %s" % (0, br.timing[0]+1e-6*br.timing[1]))
        # print("%g %s" % (-1, br.timing[-4]+1e-6*br.timing[-3]))
    if br.offset:
        print("total offsets: %d " % len(br.offset))
        # print("%d %s" % (0, br.offset[0]))
        # print("%d %s" % (-1, br.offset[-1]))
    if br.stype:
        print("total types: %d " % len(br.stype))
        # print("%d %c" % (0, br.stype[0]))
        # print("%d %c" % (-1, br.stype[-1]))
    if len(br.blobs):
        print("total blobs: %d " % (len(br.blobs)))
        # print("%d %s" % (0, br.blobs[0]))
        # print("%d %s" % (-1, br.blobs[-1]))
    nb = len(br.blobs)
    if br.timing and len(br.timing)//2 != nb:
        print("timing data %d not same size as blobs %d" % (len(br.timing)//2, nb))
    if br.offset and len(br.offset) != nb:
        print("offset data %d not same size as blobs %d" % (len(br.offset), nb))
    if br.stype and len(br.stype) != nb:
        print("type data %d not same size as blobs %d" % (len(br.stype), nb))
    #
    maxerr = 0
    result = 0
    lineno = 0
    for i in range(0, len(br.blobs)):
        lineno = 0
        try:
            bl = br.blobs[i]
            if br.stype and i < len(br.stype):
                t = br.stype[i]
            else:
                # halfassed check for json string
                if bl[0] == '{' and bl[-1] == '}':
                    t = 'j'
                else:
                    t = 's'
            if br.timing:
                try:
                    ts = (br.timing[2*i], br.timing[2*i +1])
                except:
                    ts = (int(br.filetime), 0)
            else:
                    ts = (int(br.filetime), 0)
            b = Blob(i, bl, ts, t)
            result = pf.add(b)
            if result > maxerr:
                maxerr = result
        except Exception as error:
            print("Unable to process blob ", str(i), " ", type(error).__name__, "-", error)
            traceback.print_exception(error)
            print("Blob: ",bl)
    # relocate file
    if maxerr < 2:
        pf.move_files(fn, completed_dir, "DONE") ; # either completed or unpostable
    else:
        pf.move_files(fn, retry_dir, "RETRY")
    pf.finish()
    return maxerr
 

readline0()

ldms_blob_file_to_http_post.readline0	(	typing.Union[typing.BinaryIO, int]	file_ = sys.stdin.buffer,
		bytes	separator = b'\0',
		int	blocksize = 2 ** 16
	)

Instantiate Readline0 class and yield what we get back.

file_ defaults to sys.stdin, separator defaults to a null, and blocksize defaults to 64K.

def readline0(file_: typing.Union[typing.BinaryIO, int] = sys.stdin.buffer, separator: bytes = b'\0', blocksize: int = 2 ** 16):
    # pylint: disable=W1401
    # W1401: We really do want a null byte
    """
    Instantiate Readline0 class and yield what we get back.
 
    file_ defaults to sys.stdin, separator defaults to a null, and blocksize defaults to 64K.
    """
    readline0_obj = Readline0(file_, separator, blocksize)
    # FIXME: This should become a yield from eventually.
    for line in readline0_obj.sequence():
        yield line