Linux pipe reconstruction

luigi

#! /usr/bin/env bash

# This script interrogates the system and constructs a JSON representation of the pipeline that it is part of.
# It is a proof-of-concept.  A more useful utility would accept a parameter that targets not the pipeline of the
# utility itself, but insted that of its parent, or grandparent, or great grandparent, or whatever. Such a tool could
# be used to sense the correct output-format based on who is listening downstream.

# Usage Example:
# cat -b | ./luigi | cat -e | jq .

# save pstree lines that follow the parent process id
f=$(mktemp)
pstree -ap | awk "/$PPID$/{y=1;next}y" > $f

# grab the whitespace-separated pipes preceeding the parent, this indicates our location in the treee
depth=$(head -n 1 $f | sed 's#|-.*$#|#')

mygeneration=$(mktemp)
# isolate just this generation:
# - ignore cousins (lines including and after the first non-depth-match)
# - dedent (remove depth-match)
# - ignore children (lines without a dash immediately following depth-match)
# - remove prefix (the dash)
sed -n "/$depth/!q;p" $f \
    | sed "s/^$depth//" \
    | sed -n '/^-/p' \
    | sed 's/^-//' > $mygeneration

pids=$(mktemp)
sed 's/.*,\([0-9]*\).*/\1/' $mygeneration > $pids

names=$(mktemp)
sed 's/\(.*\),[0-9]*\(.*\).*$/\1\2/' $mygeneration > $names

nodes=$(mktemp)
sed 's/.*,\([0-9]*\).*/\1/' $mygeneration | while read pid
do
    lsof -p $pid 2>/dev/null | grep pipe | sed 's/^.*[0-9]\([rw]\) *FIFO.* \([0-9]*\) pipe.*/\1,\2/' | paste -sd ";" - >> $nodes
done

read -r -d '' script <<'EOF'
import sys
import json

class AdjacentNodes:
    def __init__(self, nodes_str):
        self.read = None
        self.write = None
        node_list = nodes_str.split(';')
        for node_str in node_list:
            mode, node = node_str.split(',')
            if mode == 'r':
                self.read = node
            elif mode == 'w':
                self.write = node
            else:
                exit('unexpected mode: {}'.format(mode))
    def __repr__(self):
        return str(self.__dict__)

class Stage:
    def __init__(self, pid, name, node):
        self.pid = pid
        self.name = name
        self.node = node
    def __repr__(self):
        return str(self.__dict__)
    def as_dict(self):
        d = self.__dict__
        d['node'] = self.node.__dict__
        return d

pids = sys.argv[1].split('\n')
names = sys.argv[2].split('\n')
nodes = sys.argv[3].split('\n')

neighbors = [AdjacentNodes(pid_nodes) for pid_nodes in nodes]
unsorted = [Stage(x, y, z) for (x, y, z) in (zip(pids, names, neighbors))]

sorted = []

def print_pipeline(stage_list, file=sys.stdout):
    print(json.dumps([stage.as_dict() for stage in stage_list]), file=file)

def print_status(file=sys.stdout):
    print("Unsorted: ", end='', file=file)
    print_pipeline(unsorted, file=file)
    print("  Sorted: ", end='', file=file)
    print_pipeline(sorted, file=file)

for stage in unsorted:
    if not stage.node.read and stage.node.write:
        sorted.append(stage)
        unsorted.remove(stage)
if len(sorted) != 1:
    print_status(file=sys.stderr)
    sys.exit("couldn't find unique pipeline start")

while len(unsorted) != 1:
    found = None
    for stage in unsorted:
        if stage.node.read == sorted[-1].node.write:
            found = stage
            sorted.append(stage)
            unsorted.remove(stage)
            break
    if not found:
        print_status(file=sys.stderr)
        sys.exit("couldn't find a stage following {}".format(stage))

stage = unsorted[-1]
if stage.node.read == sorted[-1].node.write and not stage.node.write:
    sorted.append(stage)
else:
    print_status(file=sys.stderr)
    sys.exit("couldn't find unique pipeline end")

print_pipeline(sorted)
EOF

python3 -c "$script" "$(cat $pids)" "$(cat $names)" "$(cat $nodes)"