Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
generates a dot file from to Google Cloud Platform service dependencies
""""
generates a dot file from the output of `google services list --available --format json`.
it only draws services that are referenced as a dependency.
"""
import sys
from re import sub
import json
from typing import Dict, List
ServiceDictionary = Dict[str, "Service"]
all_services: ServiceDictionary = {}
class Service(object):
def __init__(self, s: dict):
super(Service, self).__init__()
self.config = s.get("config", {})
self.dependency_config = s.get("dependencyConfig", {})
self.service_accounts = s.get("serviceAccounts", [])
self.enabled = s.get("enabled")
@property
def name(self):
return self.config.get("name")
@property
def short_name(self):
return self.name.replace(".googleapis.com", "")
@property
def dot_id(self):
return sub(r"\W", "_", self.short_name)
@property
def title(self):
return self.config.get("title", self.name)
@property
def depends_on(self) -> List["Service"]:
global all_services
return list(
map(lambda n: all_services[n], self.dependency_config.get("dependsOn", []))
)
@property
def directly_depends_on(self) -> List["Service"]:
global all_services
return list(
map(
lambda n: all_services[n],
self.dependency_config.get("directlyDependsOn", []),
)
)
@staticmethod
def read_all_services(f):
global all_services
all_services = {s.name: s for s in map(lambda s: Service(s), json.load(f))}
Service.add_missing_services()
@staticmethod
def add_missing_services():
global all_services
to_add = {}
for s in all_services.values():
for d in s.dependency_config.get("directlyDependsOn", []):
if d not in all_services:
to_add[d] = Service(
{
"config": {"name": d, "title": "(unlisted)"},
"dependencyConfig": {"directlyRequiredBy": [s.name]},
}
)
all_services.update(to_add)
@staticmethod
def generate_dot():
global all_services
sys.stdout.write("digraph google_services {\n")
for _, s in all_services.items():
if s.dependency_config.get("directlyRequiredBy"):
sys.stdout.write(f' {s.dot_id} [label="{s.name}\\n{s.title}"];\n')
for _, s in all_services.items():
for d in s.directly_depends_on:
sys.stdout.write(f" {s.dot_id} -> {d.dot_id};\n")
sys.stdout.write("}\n")
def dot_id(s: str) -> str:
return sub(r"\W", "_", s.replace(".googleapis.com", ""))
if __name__ == "__main__":
Service.read_all_services(sys.stdin)
Service.generate_dot()
#!/bin/bash
[[ $# -ne 1 ]] && echo "Usage: $(basename $0) service-name" >&2 && exit 1
# allow shorthand notation, eg: cloudfunctions
SERVICE=$1
if [[ ! $SERVICE =~ .*\.googleapis.com ]]; then
SERVICE=$1.googleapis.com
fi
# cache the available services for an hour (I bet they result is pretty static)
CACHE=${TMP:-/tmp}/gcloud-available-services-cache-$(date -u +%Y-%m-%d-%H)
if [[ ! -s $CACHE ]] ; then
rm -f ${TMP:-/tmp}/gcloud-available-services-cache-$(date -u +%Y-)*
gcloud services list --available --format json > $CACHE
fi
jq --arg service "$SERVICE" \
'map(select(.config.name == $service)|
{
name: .config.name,
dependsOn: .dependencyConfig.dependsOn
}
)| .[]' $CACHE
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.