whentimeslows/website
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Consolidate YAML files [part-7] (#9262)

Browse Source 
* Consolidate YAML files [part-7]

This PR relocates YAML files used by Job/CronJob examples.

* Update examples_test.go
This commit is contained in:
Qiming
2018-07-03 04:54:18 +08:00
committed by k8s-ci-robot
parent 3a0c618734
commit 48897cc47d
17 changed files with 39 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
content/en/docs/tasks/job/_index.md
Executable → Regular
View File
content/en/docs/tasks/job/automated-tasks-with-cron-jobs.md
Executable → Regular
+1 -1
View File
@@ -43,7 +43,7 @@ component.
Cron jobs require a config file.
This example cron job config `.spec` file prints the current time and a hello message every minute:
{{< code file="cronjob.yaml" >}}
{{< codenew file="application/job/cronjob.yaml" >}}
Run the example cron job by downloading the example file and then running this command:
content/en/docs/tasks/job/coarse-parallel-processing-work-queue.md
+4 -4
View File
@@ -181,13 +181,13 @@ We will use the `amqp-consume` utility to read the message
from the queue and run our actual program. Here is a very simple
example program:
{{< code language="python" file="coarse-parallel-processing-work-queue/worker.py" >}}
{{< codenew language="python" file="application/job/rabbitmq/worker.py" >}}
Now, build an image. If you are working in the source
tree, then change directory to `examples/job/work-queue-1`.
Otherwise, make a temporary directory, change to it,
download the [Dockerfile](Dockerfile?raw=true),
and [worker.py](worker.py?raw=true). In either case,
download the [Dockerfile](/examples/application/job/rabbitmq/Dockerfile),
and [worker.py](/examples/application/job/rabbitmq/worker.py). In either case,
build the image with this command:
```shell
@@ -219,7 +219,7 @@ Here is a job definition. You'll need to make a copy of the Job and edit the
image to match the name you used, and call it `./job.yaml`.
{{< code file="coarse-parallel-processing-work-queue/job.yaml" >}}
{{< codenew file="application/job/rabbitmq/job.yaml" >}}
In this example, each pod works on one item from the queue and then exits.
So, the completion count of the Job corresponds to the number of work items
content/en/docs/tasks/job/coarse-parallel-processing-work-queue/Dockerfile
-10
View File
@@ -1,10 +0,0 @@
# Specify BROKER_URL and QUEUE when running
FROM ubuntu:14.04
RUN apt-get update && \
apt-get install -y curl ca-certificates amqp-tools python \
--no-install-recommends \
&& rm -rf /var/lib/apt/lists/*
COPY ./worker.py /worker.py
CMD /usr/bin/amqp-consume --url=$BROKER_URL -q $QUEUE -c 1 /worker.py
content/en/docs/tasks/job/coarse-parallel-processing-work-queue/job.yaml
-20
View File
@@ -1,20 +0,0 @@
apiVersion: batch/v1
kind: Job
metadata:
name: job-wq-1
spec:
completions: 8
parallelism: 2
template:
metadata:
name: job-wq-1
spec:
containers:
- name: c
image: gcr.io/<project>/job-wq-1
env:
- name: BROKER_URL
value: amqp://guest:guest@rabbitmq-service:5672
- name: QUEUE
value: job1
restartPolicy: OnFailure
content/en/docs/tasks/job/coarse-parallel-processing-work-queue/worker.py
-7
View File
@@ -1,7 +0,0 @@
#!/usr/bin/env python
# Just prints standard out and sleeps for 10 seconds.
import sys
import time
print("Processing " + sys.stdin.lines())
time.sleep(10)
content/en/docs/tasks/job/cronjob.yaml
-18
View File
@@ -1,18 +0,0 @@
apiVersion: batch/v1beta1
kind: CronJob
metadata:
name: hello
spec:
schedule: "*/1 * * * *"
jobTemplate:
spec:
template:
spec:
containers:
- name: hello
image: busybox
args:
- /bin/sh
- -c
- date; echo Hello from the Kubernetes cluster
restartPolicy: OnFailure
content/en/docs/tasks/job/fine-parallel-processing-work-queue.md
+24 -12
View File
@@ -12,7 +12,6 @@ worker processes in a given pod.
In this example, as each pod is created, it picks up one unit of work
from a task queue, processes it, and repeats until the end of the queue is reached.
Here is an overview of the steps in this example:
1. **Start a storage service to hold the work queue.** In this example, we use Redis to store
@@ -50,16 +49,27 @@ For this example, for simplicity, we will start a single instance of Redis.
See the [Redis Example](https://github.com/kubernetes/examples/tree/master/guestbook) for an example
of deploying Redis scalably and redundantly.
Start a temporary Pod running Redis and a service so we can find it.
If you are working from the website source tree, you can go to the following
directory and start a temporary Pod running Redis and a service so we can find it.
```shell
$ kubectl create -f docs/tasks/job/fine-parallel-processing-work-queue/redis-pod.yaml
$ cd content/en/examples/application/job/redis
$ kubectl create -f ./redis-pod.yaml
pod "redis-master" created
$ kubectl create -f docs/tasks/job/fine-parallel-processing-work-queue/redis-service.yaml
$ kubectl create -f ./redis-service.yaml
service "redis" created
```
If you're not working from the source tree, you could also download [`redis-pod.yaml`](redis-pod.yaml?raw=true) and [`redis-service.yaml`](redis-service.yaml?raw=true) directly.
If you're not working from the source tree, you could also download the following
files directly:
- [`redis-pod.yaml`](/examples/application/job/redis/redis-pod.yaml)
- [`redis-service.yaml`](/examples/application/job/redis/redis-service.yaml)
- [`Dockerfile`](/examples/application/job/redis/Dockerfile)
- [`job.yaml`](/examples/application/job/redis/job.yaml)
- [`rediswq.py`](/examples/application/job/redis/rediswq.py)
- [`worker.py`](/examples/application/job/redis/worker.py)
## Filling the Queue with tasks
@@ -122,17 +132,19 @@ We will use a python worker program with a redis client to read
the messages from the message queue.
A simple Redis work queue client library is provided,
called rediswq.py ([Download](rediswq.py?raw=true)).
called rediswq.py ([Download](/examples/application/job/redis/rediswq.py)).
The "worker" program in each Pod of the Job uses the work queue
client library to get work. Here it is:
{{< code language="python" file="fine-parallel-processing-work-queue/worker.py" >}}
{{< codenew language="python" file="application/job/redis/worker.py" >}}
If you are working from the source tree,
change directory to the `docs/tasks/job/fine-parallel-processing-work-queue/` directory.
Otherwise, download [`worker.py`](worker.py?raw=true), [`rediswq.py`](rediswq.py?raw=true), and [`Dockerfile`](Dockerfile?raw=true)
using above links. Then build the image:
If you are working from the source tree, change directory to the
`content/en/examples/application/job/redis/` directory.
Otherwise, download [`worker.py`](/examples/application/job/redis/worker.py),
[`rediswq.py`](/examples/application/job/redis/rediswq.py), and
[`Dockerfile`](/examples/application/job/redis/Dockerfile) files, then build
the image:
```shell
docker build -t job-wq-2 .
@@ -166,7 +178,7 @@ gcloud docker -- push gcr.io/<project>/job-wq-2
Here is the job definition:
{{< code file="fine-parallel-processing-work-queue/job.yaml" >}}
{{< codenew file="application/job/redis/job.yaml" >}}
Be sure to edit the job template to
change `gcr.io/myproject` to your own path.
content/en/docs/tasks/job/fine-parallel-processing-work-queue/Dockerfile
-6
View File
@@ -1,6 +0,0 @@
FROM python
RUN pip install redis
COPY ./worker.py /worker.py
COPY ./rediswq.py /rediswq.py
CMD python worker.py
content/en/docs/tasks/job/fine-parallel-processing-work-queue/job.yaml
-14
View File
@@ -1,14 +0,0 @@
apiVersion: batch/v1
kind: Job
metadata:
name: job-wq-2
spec:
parallelism: 2
template:
metadata:
name: job-wq-2
spec:
containers:
- name: c
image: gcr.io/myproject/job-wq-2
restartPolicy: OnFailure
content/en/docs/tasks/job/fine-parallel-processing-work-queue/redis-pod.yaml
-15
View File
@@ -1,15 +0,0 @@
apiVersion: v1
kind: Pod
metadata:
name: redis-master
labels:
app: redis
spec:
containers:
- name: master
image: redis
env:
- name: MASTER
value: "true"
ports:
- containerPort: 6379
content/en/docs/tasks/job/fine-parallel-processing-work-queue/redis-service.yaml
-10
View File
@@ -1,10 +0,0 @@
apiVersion: v1
kind: Service
metadata:
name: redis
spec:
ports:
- port: 6379
targetPort: 6379
selector:
app: redis
content/en/docs/tasks/job/fine-parallel-processing-work-queue/rediswq.py
-130
View File
@@ -1,130 +0,0 @@
#!/usr/bin/env python
# Based on http://peter-hoffmann.com/2012/python-simple-queue-redis-queue.html
# and the suggestion in the redis documentation for RPOPLPUSH, at
# http://redis.io/commands/rpoplpush, which suggests how to implement a work-queue.
import redis
import uuid
import hashlib
class RedisWQ(object):
"""Simple Finite Work Queue with Redis Backend
This work queue is finite: as long as no more work is added
after workers start, the workers can detect when the queue
is completely empty.
The items in the work queue are assumed to have unique values.
This object is not intended to be used by multiple threads
concurrently.
"""
def __init__(self, name, **redis_kwargs):
"""The default connection parameters are: host='localhost', port=6379, db=0
The work queue is identified by "name". The library may create other
keys with "name" as a prefix.
"""
self._db = redis.StrictRedis(**redis_kwargs)
# The session ID will uniquely identify this "worker".
self._session = str(uuid.uuid4())
# Work queue is implemented as two queues: main, and processing.
# Work is initially in main, and moved to processing when a client picks it up.
self._main_q_key = name
self._processing_q_key = name + ":processing"
self._lease_key_prefix = name + ":leased_by_session:"
def sessionID(self):
"""Return the ID for this session."""
return self._session
def _main_qsize(self):
"""Return the size of the main queue."""
return self._db.llen(self._main_q_key)
def _processing_qsize(self):
"""Return the size of the main queue."""
return self._db.llen(self._processing_q_key)
def empty(self):
"""Return True if the queue is empty, including work being done, False otherwise.
False does not necessarily mean that there is work available to work on right now,
"""
return self._main_qsize() == 0 and self._processing_qsize() == 0
# TODO: implement this
# def check_expired_leases(self):
# """Return to the work queueReturn True if the queue is empty, False otherwise."""
# # Processing list should not be _too_ long since it is approximately as long
# # as the number of active and recently active workers.
# processing = self._db.lrange(self._processing_q_key, 0, -1)
# for item in processing:
# # If the lease key is not present for an item (it expired or was
# # never created because the client crashed before creating it)
# # then move the item back to the main queue so others can work on it.
# if not self._lease_exists(item):
# TODO: transactionally move the key from processing queue to
# to main queue, while detecting if a new lease is created
# or if either queue is modified.
def _itemkey(self, item):
"""Returns a string that uniquely identifies an item (bytes)."""
return hashlib.sha224(item).hexdigest()
def _lease_exists(self, item):
"""True if a lease on 'item' exists."""
return self._db.exists(self._lease_key_prefix + self._itemkey(item))
def lease(self, lease_secs=60, block=True, timeout=None):
"""Begin working on an item the work queue.
Lease the item for lease_secs. After that time, other
workers may consider this client to have crashed or stalled
and pick up the item instead.
If optional args block is true and timeout is None (the default), block
if necessary until an item is available."""
if block:
item = self._db.brpoplpush(self._main_q_key, self._processing_q_key, timeout=timeout)
else:
item = self._db.rpoplpush(self._main_q_key, self._processing_q_key)
if item:
# Record that we (this session id) are working on a key. Expire that
# note after the lease timeout.
# Note: if we crash at this line of the program, then GC will see no lease
# for this item a later return it to the main queue.
itemkey = self._itemkey(item)
self._db.setex(self._lease_key_prefix + itemkey, lease_secs, self._session)
return item
def complete(self, value):
"""Complete working on the item with 'value'.
If the lease expired, the item may not have completed, and some
other worker may have picked it up. There is no indication
of what happened.
"""
self._db.lrem(self._processing_q_key, 0, value)
# If we crash here, then the GC code will try to move the value, but it will
# not be here, which is fine. So this does not need to be a transaction.
itemkey = self._itemkey(value)
self._db.delete(self._lease_key_prefix + itemkey, self._session)
# TODO: add functions to clean up all keys associated with "name" when
# processing is complete.
# TODO: add a function to add an item to the queue. Atomically
# check if the queue is empty and if so fail to add the item
# since other workers might think work is done and be in the process
# of exiting.
# TODO(etune): move to my own github for hosting, e.g. github.com/erictune/rediswq-py and
# make it so it can be pip installed by anyone (see
# http://stackoverflow.com/questions/8247605/configuring-so-that-pip-install-can-work-from-github)
# TODO(etune): finish code to GC expired leases, and call periodically
# e.g. each time lease times out.
content/en/docs/tasks/job/fine-parallel-processing-work-queue/worker.py
-23
View File
@@ -1,23 +0,0 @@
#!/usr/bin/env python
import time
import rediswq
host="redis"
# Uncomment next two lines if you do not have Kube-DNS working.
# import os
# host = os.getenv("REDIS_SERVICE_HOST")
q = rediswq.RedisWQ(name="job2", host="redis")
print("Worker with sessionID: " + q.sessionID())
print("Initial queue state: empty=" + str(q.empty()))
while not q.empty():
item = q.lease(lease_secs=10, block=True, timeout=2)
if item is not None:
itemstr = item.decode("utf=8")
print("Working on " + itemstr)
time.sleep(10) # Put your actual work here instead of sleep.
q.complete(item)
else:
print("Waiting for work")
print("Queue empty, exiting")
content/en/docs/tasks/job/job.yaml
-18
View File
@@ -1,18 +0,0 @@
apiVersion: batch/v1
kind: Job
metadata:
name: process-item-$ITEM
labels:
jobgroup: jobexample
spec:
template:
metadata:
name: jobexample
labels:
jobgroup: jobexample
spec:
containers:
- name: c
image: busybox
command: ["sh", "-c", "echo Processing item $ITEM && sleep 5"]
restartPolicy: Never
content/en/docs/tasks/job/parallel-processing-expansion.md
+4 -4
View File
@@ -18,9 +18,9 @@ non-parallel, use of [Jobs](/docs/concepts/jobs/run-to-completion-finite-workloa
## Basic Template Expansion
First, download the following template of a job to a file called `job.yaml`
First, download the following template of a job to a file called `job-tmpl.yaml`
{{< code file="job.yaml" >}}
{{< codenew file="application/job/job-tmpl.yaml" >}}
Unlike a *pod template*, our *job template* is not a Kubernetes API type. It is just
a yaml representation of a Job object that has some placeholders that need to be filled
@@ -47,7 +47,7 @@ Next, expand the template into multiple files, one for each item to be processed
$ mkdir ./jobs
$ for i in apple banana cherry
do
cat job.yaml | sed "s/\$ITEM/$i/" > ./jobs/job-$i.yaml
cat job-tmpl.yaml | sed "s/\$ITEM/$i/" > ./jobs/job-$i.yaml
done
```
@@ -195,4 +195,4 @@ If you have a large number of job objects, you may find that:
In this case, you can consider one of the
other [job patterns](/docs/concepts/jobs/run-to-completion-finite-workloads/#job-patterns).
{{% /capture %}}
{{% /capture %}}