Pipeline worker cloud

Source: 14_cloud_worker_pipeline

Lit un fichier texte local, l’envoie à un runner cloud pour un traitement de comptage de mots (un compte par ligne), et écrit les résultats localement. Le runner cloud est libéré explicitement une fois le pipeline terminé.

Exécution


melodium run 14_cloud_worker_pipeline/Compo.toml \
  --api_token "my-api-token" \
  --input     data.txt \
  --output    word_counts.txt

Note

api_token authentifie ici auprès d’une API Mélodium Services, comme Cadence.CI .


[…] info: cloud: provisioning cloud runner…
[…] info: cloud: runner provisioned, distributor connecting…
[…] info: cloud: pipeline complete

Fonctionnement

main instancie les modèles DistantEngine et DistributionEngine, puis câble le pipeline lecture/distribution/écriture :


model runner: DistantEngine(api_url=|wrap<string>("https://api-staging.melodium.tech/0.1"), api_token=|wrap<string>(api_token))
model distributor: DistributionEngine(
  treatment = "cloud_worker_pipeline/main::transform",
  version   = "0.1.0"
)

Provisionnement et verrou de lecture

distant provisionne un conteneur (512 Mo de RAM, 1 CPU, 1 Go de stockage) ; sa sortie access alimente start, et le fichier local n’est lu qu’une fois le distributeur passé à ready :


provisionRunner: distant[distant_engine=runner](
    max_duration = 600,
    memory       = 512, // Mo
    cpu          = 1000, // millicœurs
    storage      = 1024, // Mo
    edition      = _,
    arch         = _,
    volumes      = [],
    containers   = [],
    service_containers = [],
    tags         = []
)
startup.trigger -> provisionRunner.trigger,access -> distribStart.access
 
distribStart: start[distributor=distributor](params=|map([]))
 
read: readLocal(path=input)
distribStart.ready -> read.trigger

main treatment diagram Voir dans Compositeur Studio

Distribution vers le `transform` distant

dispatch encapsule distribute, sendStream et recvStream. À noter que les types envoyés et reçus diffèrent : des octets bruts partent, des chaînes de comptage reviennent :


treatment dispatch[distributor: DistributionEngine]()
  input  data:   Stream<byte>
  output result: Stream<string>
{
    trig: trigger<byte>()
    dist: distribute[distributor=distributor]()
 
    Self.data -> trig.stream,start -> dist.trigger
 
    sendData:   sendStream<byte>[distributor=distributor](name="data")
    recvResult: recvStream<string>[distributor=distributor](name="result")
 
    dist.distribution_id -> sendData.distribution_id
    dist.distribution_id -> recvResult.distribution_id
 
    Self.data       -> sendData.data
    recvResult.data -> Self.result
}
 
read.data -> dispatch.data,result -> write.text

Les valeurs de name ("data", "result") correspondent exactement aux noms des ports d’entrée/sortie de transform.

Nettoyage explicite

Contrairement aux autres exemples distribués où le runner tourne indéfiniment, ce pipeline appelle stop[distributor=distributor]() une fois l’écriture terminée :


distribStop: stop[distributor=distributor]()
write.completed -> distribStop.trigger

Cela libère le runner cloud et arrête la facturation. Le max_duration = 600 sur provisionRunner sert de sécurité si le nettoyage échoue.

Le traitement distant `transform`

Le modèle WordCounter embarque une fonction JavaScript qui compte les mots séparés par des espaces par ligne. Il est défini dans le même fichier que main mais n’est instancié que sur le runner distant :


model WordCounter() : JavaScriptEngine {
    code = ${{function countWords(line) {
    var s = line.trim();
    if (s.length === 0) return '0';
    return s.split(/\s+/).length.toString();
}
}}
}
 
treatment transform()
  model counter: WordCounter()
  input  data:   Stream<byte>
  output result: Stream<string>
{
    decode()
    wrapStr:      fromString<string>()
    jsCount:      process[engine=counter](code="countWords(value)")
    unwrapResult: unwrapOr<Json>(default=|null())
    resultStr:    tryToString<Json>()
    unwrapStr:    unwrapOr<string>(default="0")
 
    Self.data -> decode.data,text -> wrapStr.value,json -> jsCount.value,result -> unwrapResult.option,value -> resultStr.value,into -> unwrapStr.option,value -> Self.result
}

decode convertit les octets en texte UTF-8, fromString<string>() enveloppe chaque ligne en valeur Json pour process, et la chaîne unwrapOr / tryToString extrait ensuite la chaîne de comptage, avec "0" comme valeur par défaut en cas d’erreur.

Dépendances


[dependencies]
std        = "0.10.1"  # flux de base, journalisation, structures de données
fs         = "0.10.1"  # lecture/écriture de fichiers locaux
encoding   = "0.10.1"  # encodage / décodage UTF-8
javascript = "0.10.1"  # moteur JavaScript embarqué
json       = "0.10.1"  # parsing et sérialisation JSON
work       = "0.10.1"  # provisionnement de runners cloud
distrib    = "0.10.1"  # distribution de flux entre runners