Merge pull request #87 from AlekSi/patch-1

Make go vet happy

.gitignore

@@ -1 +1,4 @@
 *~
+
+# Pycharm IDE
+.idea

.travis.yml

@@ -0,0 +1,20 @@
+sudo: required
+dist: trusty
+
+language: go
+go:
+    - 1.7
+    - tip
+
+install:
+   - go get github.com/Masterminds/glide
+   - curl -L https://github.com/coreos/etcd/releases/download/v3.0.15/etcd-v3.0.15-linux-amd64.tar.gz -o /tmp/etcd-v3.0.15-linux-amd64.tar.gz && mkdir -p /tmp/test-etcd && tar xzvf /tmp/etcd-v3.0.15-linux-amd64.tar.gz -C /tmp/test-etcd --strip-components=1
+
+script:
+   - cd ${TRAVIS_BUILD_DIR}
+   - glide install
+   - ./fission-bundle/build.sh
+   - hack/verify-gofmt.sh
+   - /tmp/test-etcd/etcd &
+   - go test -v -i $(go list ./... | grep -v '/vendor/')
+   - go test -v $(go list ./... | grep -v '/vendor/')

CODE_OF_CONDUCT.md

@@ -0,0 +1,74 @@
+# Contributor Covenant Code of Conduct
+
+## Our Pledge
+
+In the interest of fostering an open and welcoming environment, we as
+contributors and maintainers pledge to making participation in our project and
+our community a harassment-free experience for everyone, regardless of age, body
+size, disability, ethnicity, gender identity and expression, level of experience,
+nationality, personal appearance, race, religion, or sexual identity and
+orientation.
+
+## Our Standards
+
+Examples of behavior that contributes to creating a positive environment
+include:
+
+* Using welcoming and inclusive language
+* Being respectful of differing viewpoints and experiences
+* Gracefully accepting constructive criticism
+* Focusing on what is best for the community
+* Showing empathy towards other community members
+
+Examples of unacceptable behavior by participants include:
+
+* The use of sexualized language or imagery and unwelcome sexual attention or
+advances
+* Trolling, insulting/derogatory comments, and personal or political attacks
+* Public or private harassment
+* Publishing others' private information, such as a physical or electronic
+  address, without explicit permission
+* Other conduct which could reasonably be considered inappropriate in a
+  professional setting
+
+## Our Responsibilities
+
+Project maintainers are responsible for clarifying the standards of acceptable
+behavior and are expected to take appropriate and fair corrective action in
+response to any instances of unacceptable behavior.
+
+Project maintainers have the right and responsibility to remove, edit, or
+reject comments, commits, code, wiki edits, issues, and other contributions
+that are not aligned to this Code of Conduct, or to ban temporarily or
+permanently any contributor for other behaviors that they deem inappropriate,
+threatening, offensive, or harmful.
+
+## Scope
+
+This Code of Conduct applies both within project spaces and in public spaces
+when an individual is representing the project or its community. Examples of
+representing a project or community include using an official project e-mail
+address, posting via an official social media account, or acting as an appointed
+representative at an online or offline event. Representation of a project may be
+further defined and clarified by project maintainers.
+
+## Enforcement
+
+Instances of abusive, harassing, or otherwise unacceptable behavior may be
+reported by contacting the project team at soamvasani@platform9.com. All
+complaints will be reviewed and investigated and will result in a response that
+is deemed necessary and appropriate to the circumstances. The project team is
+obligated to maintain confidentiality with regard to the reporter of an incident.
+Further details of specific enforcement policies may be posted separately.
+
+Project maintainers who do not follow or enforce the Code of Conduct in good
+faith may face temporary or permanent repercussions as determined by other
+members of the project's leadership.
+
+## Attribution
+
+This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 1.4,
+available at [http://contributor-covenant.org/version/1/4][version]
+
+[homepage]: http://contributor-covenant.org
+[version]: http://contributor-covenant.org/version/1/4/

CONTRIBUTING.md

@@ -0,0 +1,46 @@
+# Contributing to Fission
+
+### Thank you!
+
+Thank you for taking the time and effort to contribute to Fission!
+
+First, please read the [code of conduct](CODE_OF_CONDUCT.md).  By
+participating, you're expected to uphold this code.
+
+### Choose something to work on
+
+The easiest way to start is to look at existing
+[issues](https://github.com/fission/fission/issues) and see if there's
+something there that you'd like to work on.
+
+Also, if you're familiar with a language that we don't support today,
+adding an environment for a new language is a good task to take on.
+Take a look at the Python environment as an example.
+
+If you're interested in working on something that doesn't have an
+existing issue, feel free to file a new one.
+
+There's plenty of stuff to do.  If you need help choosing something to
+work on, we'll be happy to suggest something depending on your
+interests.
+
+### Talk to people
+
+At any point, from thinking about contributing to merging your
+changes, please feel free to reach out to us!  We're more than happy
+to talk.
+
+If you choose a big chunk of work that involves design changes to
+Fission, please talk to us while planning the change; that way we can
+all be on the same page before you invest too much work into it.
+
+ * Drop by the [slack channel](http://slack.fission.io)
+ * Say hi on [twitter](https://twitter.com/fissionio).
+ * Email soamvasani@platform9.com
+
+### Contribution Howto
+
+ * Fork this github repo
+ * Make a branch in your fork
+ * Commit your changes to that branch
+ * Create a pull request

Documentation/Architecture.md

@@ -1,107 +1,113 @@
 
 A high level view of the internals of Fission.
 
+How it works
+============
+
+Fission is a FaaS -- users create functions (source level), register
+them with fission using a CLI, and associate functions with triggers.
+
+Fission wraps those functions into a service, and runs them on
+Kubernetes on demand.
+
+Here's an overview of the services that make up fission.
+
 Components
 ==========
 
 Language-neutral components:
 
- * Controller
+ * controller
- * Container Pool Manager
+ * poolmgr
- * Container Specializer
+ * router
- * Router
+ * kubewatcher
 
 Language-specific components:
 
- * Language Build Container
+ * Environment container
- * Language Run Container
 
 
 Controller
 ----------
 
-Function and Trigger CRUD APIs.  APIs to watch for changes are also
+The controller contains CRUD APIs for functions, http triggers,
-included (useful for other components that cache state).
+environments, Kubernetes event watches.  This is the component that
-
+the client talks to.
-See api/swagger.json for API details.
 
 This is the only stateful component.  It needs to be configured with a
 URL to an etcd cluster and a path to a persistent volume.  The volume
-will be used to store the functions' source code.
+is used to store the functions' source code.  Etcd is used as the DB.
 
-Etcd is used as the DB.
+[Work to extend to other storage backends is planned, see issue #83.]
 
+Pool Manager 
+------------
 
-Container Pool Manager 
+Poolmgr manages pools of generic containers and function containers.
-----------------------
 
-Manage pool of generic containers.
+It has a simple API; both these endpoints are called by the router.
 
-Probably use K8s RCs.  Can we Use labels to move pods from one rc to
+* GetFunctionService takes function metadata and returns the address
-another?  What about jobs?
+  of a service.
+  
+* TapService lets poolmgr know a service is being used; if it's not
+  called for a few minutes the pod(s) backing the service are killed.
 
-Container Specializer
+Poolmgr watches the controller API and eagerly creates generic pools
----------------------
+for environments.  It uses Kubernetes deployments to do that.  The
+environment container runs in a pod with the 'fetcher' container.
+Fetcher is a very simple utility that downloads a URL sent to it and
+saves it at a configured location.
 
-Inputs: a running generic language run container, a user function,
+GetFunctionService "specializes" a pod.  The implementation chooses a
-optionally an http trigger URL.
+pod from the pool, relabels it to "orphan" the pod from the
-
+deployment, invokes fetcher to copy the function into the pod, and
-Calls Language Run Container and sets up Router to point to it.
+hits the the specialize endpoint on the environment container.  This
+causes the function to be loaded.  The pod is now specific to that
+function.
 
+This function pod is cached; it's cleaned up if it's unused for a few
+minutes.
 
 Router
 ------
 
-- Cache trigger -> container instance mapping; implement cache miss and expiration.
+The router forwards HTTP requests to function pods.  If there's no
+running service for a function, it requests one from poolmgr, while
+holding on to the request; when the function's service is ready it
+forwards the request.
 
-- Invoke Specializer, setup up k8s API 
+The router is stateless and can be scaled up if needed, according to
+load.
 
-- Forward requests
+Kubewatcher
+-----------
 
-The Router is stateless -- it can be scaled or killed at any time.
+Kubewatcher watches the Kubernetes API and invokes functions
+associated with watches, sending the watch event to the function.
 
-There's a lot of functionality overlap with K8S Ingress Controllers.
+The controller keeps track of user's requested watches and associated
-We should clearly use Ingress and Ingress Controllers in some way.
+functions.  Kubewatcher watches the API based on these requests; when
-It's not exactly clear at the moment how -- should make a whole new
+a watch event occurs, it serializes the object and calls the function
-Ingress Controller perhaps based on the contrib/nginx
+via the router.
 
+While a few simple retries are done, there isn't yet a reliable
+message bus between Kubewatcher and the function.  Work for this is
+tracked in issue #64.
 
-Autoscaler
+Environment Container
-----------
+---------------------
-
-This autoscales the language run containers that are backing a
-trigger.
-
-What metrics this is based on is TBD.
-
-- Number of requests/sec
-- "Backlog" -- number of outstanding requests not yet started -- how to measure this?
-- Change in turn-around time?
-
-
-Language Build Container
-------------------------
-
-* The Language Build Container is a container that is invoked for a
-  build.  It takes one user-created function and outputs something
-  that can be run by the corresponding Language Run Container.
-
-* The Build Container must implement the Language Build Container
-  interface.
-
-
-Language Run Container
-----------------------
-
-* The Language Run Container is the container in which user functions
-  run.
-
-* The Run Container is started without the user function.  It must
-  start as a "Generic Container".  It must implement the
-  "specialization interface".  In short, it must implement an HTTP
-  server that can receive a piece of code, verify its signature, and
-  map it to an HTTP endpoint.  See
-  Documentation/specs/LanguageRunContainerSpec.md for details.
 
+Environment containers run user-defined functions.  Environment
+containers are language specific.  They must contain an HTTP server
+and a loader for functions.
 
+Poolmgr deploys the environment container into a pod with fetcher
+(fetcher is a simple utility that can fetch an HTTP url to a file at a
+configured location).  This pod forms a "generic pod", because it can
+be loaded with any function.
 
+When poolmgr needs to create a service for a function, it calls
+fetcher to fetch the function.  Fetcher downloads the function into a
+volume shared between fetcher and this environment container.  Poolmgr
+then requests the container to load the function.

Documentation/ProgrammingModel.md

@@ -1,71 +0,0 @@
-Programming Model
-=================
-
-This document describes the programming model for fission functions.  
-
-See TERMINOLOGY for definitions of terms, such as _function_,
-_instance_, and _trigger_.
-
-
-Idempotency
-===========
-
-Fission assumes that functions are idempotent.  Functions whose instances
-die without a result are restarted, upto a certain restart limit.
-
-
-Time Limits
-===========
-
-By default, there is no time limit on fission functions.  
-
-Idle running instances may be killed at any time (usually after the
-default idle timeout of 10 minutes, but this is configurable).
-
-
-Mapping
-=======
-
-This section specifies the mapping between: (1) HTTP and other
-triggers, and (2) function parameters, return values, and exceptions.
-
-In other words, this section describes how a function should be called
-based on a given trigger, and how the functions behaviour affects the
-result returned from the trigger.
-
-The HTTP Trigger
-----------------
-
-NodeJS
-------
-
-NodeJS functions are called with a context object.
-
-context.request contains the nodeJS Request object.
-
-In addition,
-
-context.queryString contains the parsed querystring
-
-context.body contains the parsed body
-
-context.status sets the HTTP response status code.  If context.status
-is an invalid HTTP status, then the HTTP status is set to 500.
-
-Exceptions result in a HTTP 500 error.
-
-
-
-
-
-XXX Should we have a "context.done()"?  What are the pros and cons?
-
-XXX Who handles serialization and deserialization?  Should we do json
-    automatically based on content-type and accept headers?
-
-
-Python
-------
-
-Same as node, pretty much.
-

Documentation/TERMINOLOGY

@@ -1,22 +1,15 @@
-
+_Function_: A fission function is something that's mapped to a
-_Function_: A _function_ is the smallest unit of a program in Fission.
+_trigger_ and run on demand.  Though we call it a "function", this is
-It's a program with an function as an entry point -- it doesn't have
+a bit imprecise, since it's actually a module with an function as an
-to be just one function.
+entry point -- it doesn't have to be just one function.
 
 _Trigger_: Triggers are what cause functions to be called.  For
-example the HTTP trigger causes functions to be called on HTTP
+example, an HTTP trigger causes functions to be called on HTTP
-requests.
+requests.  Kubernetes Watch triggers cause functions to be called when
+a Kubernetes watch changes.  Future triggers will include message
+queues, timers, storage systems, etc.
 
-_Instance_: Fission creates and runs servers containing these
+_Environments_: Environments are the language-specific parts of
-functions.  The running server is called an _instance_.  You normally
+Fission.  Environment containers wrap the user's function and present
-don't need to worry about instances: their management is completely
+a common interface to the rest of the fission framework.
-automatic and transparent to you.  However, certain configuration
-settings allow you to tweak the behaviour of instances -- for example,
-you can configure the strategy for destroying unused instances.
-
-_Generic Instance_: An instance with no user code in it.  The process
-of including the user code to a Generic Instance is _Specialization_,
-and the resulting instance is a _Specialized Instance_.
-
-_Specialized Instance_: See _General Instance_.
 

Documentation/UserManual.md

@@ -1,15 +0,0 @@
-
-Fission is functions-as-a-service for Kubernetes.
-
-You write some functions, and setup some way for them to be invoked
-("triggers").  Triggers can be HTTP requests (HTTP Triggers) or timers
-(Time Triggers).
-
-Topics
-======
-
-Testing
--------
-
-Function Updates
-----------------

Documentation/wip/ProgrammingModel.md

@@ -0,0 +1,24 @@
+# Programming Model
+
+This document describes the programming model for fission functions.  
+
+## Time Limits
+
+By default, there is no time limit on fission functions.  
+
+Idle running instances may be killed at any time (usually after the
+default idle timeout of 10 minutes, but this is configurable).
+
+
+## HTTP Triggers
+
+Functions triggered over HTTP receive the HTTP request object in the
+context.  The request's query string, POST body, etc. can be retrieved
+from this object.  The interface is language-specific: see [TODO] for
+documentation on the context object in each environment.
+
+
+## Kubernetes Watch Event Triggers
+
+Kubernetes watches can be used to trigger functions.  These functions
+receive the Kubernetes watch.Event object in JSON-serialized form.

Documentation/wip/environments-v2.md

@@ -0,0 +1,167 @@
+# Fission Environments Redesign
+
+As Fission supports more languages and reaches a wider set of use
+cases, it's time to ask how well the current Environments design is
+holding up.
+
+
+## Environments V1: What we learned
+
+Environments V1 is very simple idea: an environments is one Docker
+image with an HTTP server + dynamic loader for that language; it's run
+in a pod with a language-agnostic sidecar (fetcher) that downloads and
+saves the function into a volume shared with the language-specific
+container.
+
+### Pros:
+
+* Abstracted away images.
+
+* Very fast cold start
+
+* No image registry to manage (neither for the user nor for fission
+  implementation)
+
+* Relatively small amount of language specific code. (python env is <
+  100 lines)
+
+### Cons:
+
+* Doesn't work well for compiled languages
+
+* Users have to rebuild the image to add dependencies
+
+* Only one file supported
+
+* Errors in loading are not surfaced properly.  It is especially
+  annoying to wait until runtime to see a syntax error that could have
+  been caught on function upload.
+
+* Not great for a large code base
+
+* Some people want to operate at the image level but still get the
+  on-demand execution semantics of FaaS.  This is a cost-optimization
+  use case.
+
+
+### Discussion
+
+Early feedback shows that almost evey user ends up rebuilding images
+to add some dependecies.  Some sort of automated dependecy resolution
+would be very nice to have and improve the development workflow.  In
+other words, just attach a package.json(nodejs) or
+requirements.txt(python) with a function, and fission will do the
+rest.  There's also the possiblity of supporting buildpacks (simple
+zipfiles), a la AWS Lambda.
+
+Though we can support compiled languages by doing the compilation
+inside the cold-start, that's not a great solution because: (a)
+compile errors would be reported at runtime, and (b) because the
+overhead of compilation doesn't really need to be inside the
+cold-start latency.
+
+Non-trivial functions will need multiple files.  That also helps for
+common code across functions.  So we need a way for the user to define
+a function as a collection of code with an entry point.
+
+Finally, Docker images remain the most flexible way to package an app.
+Today, users can always rebuild an environment image to include
+anything they want.  But those images must still run a server that
+implements fission-environment interface (i.e. the specialize
+endpoint).  So perhaps there could be a way for users to say "don't
+use environments, I've already packaged up my function, here it is".
+
+
+## Environment V2 Requirements
+
+Roughly in order of priority:
+
+0. Retain the simplicity of the simple use cases.  First user
+   experience shoud remain trivial -- write a function, map a URL,
+   done.
+
+1. Support compiled languages. Support error reporting on function
+   upload rather than cold start.
+
+2. Support functions as a collection of files rather than just one
+   file.
+
+3. Support automated environment-specific dependecy resolution.
+
+   (#3 may end up having the same solution as #1.  You could think of
+   gathering deps as a "compilation" of package.json,
+   requirements.txt, etc.)
+
+4. Support functions as images.
+
+
+### User stories
+
+#### Compiled language
+
+User writes a function in Go.
+
+```
+      $ fission function create --code blah.go
+
+      <compilation errors>
+
+      <user edits file>
+      $ $EDITOR blah.go 
+      <fixes errors>
+
+      $ fission function update --code blah.go
+      <success>
+```
+
+(Or perhaps we could have a `fission function check --env x --code y`
+which just does compilation, without creating a function object?
+Useful for integration into IDEs.  Basically, just an on-demand
+builder. Useful when you don't wanna setup anything on your laptop.)
+
+This same user story applies to interpreted languages too, where the
+"compilation" step can be used to check for syntax errors.
+
+
+#### Collections of files
+
+We should probably have a manifest in YAML/JSON/etc syntax for
+specifying a function.  We could also use that YAML to let users
+specify the function's environment, resource requirements, etc.
+
+```
+        $ fission funcion create -f blah.yaml
+
+        $ cat blah.yaml
+        type: Function
+        metadata:
+           name: ...
+        environment: ...
+        files:
+        - foo.py
+        - bar.py
+        - baz/*.py              
+```     
+
+The yaml file could specify a list of files.  The fission client would
+deal with packaging up this set of files and uploading the package.
+
+
+#### Handling Dependencies
+
+A manifest could point at a function's dependency spec.  The
+Environment's "builder" container could then fetch these deps.
+
+So a NodeJS function manifest could contain a reference to
+package.json.  A "builder" container in the NodeJS environment would
+then run npm on that function.
+
+```     $ cat func.yaml
+        type: Function
+        metadata: ...
+        environment: ...
+        dependencies: package.json
+
+        $ fission function create -f func.yaml
+```
+

INSTALL.md

@@ -0,0 +1,112 @@
+  * [Running Fission on your Cluster](#running-fission-on-your-cluster)
+     * [Setup Kubernetes](#setup-kubernetes)
+        * [Mac](#install-and-start-kubernetes-on-osx)
+        * [Linux](#or-install-and-start-kubernetes-on-linux)
+     * [Verify access to the cluster](#verify-access-to-the-cluster)
+     * [Get and Run Fission: Minikube or Local cluster](#get-and-run-fission-minikube-or-local-cluster)
+     * [Get and Run Fission: GKE or other Cloud](#get-and-run-fission-gke-or-other-cloud)
+     * [Install the client CLI](#install-the-client-cli)
+     * [Run an example](#run-an-example)
+
+## Running Fission on your Cluster
+
+### Setup Kubernetes
+
+You can install Kubernetes on your laptop with [minikube](https://github.com/kubernetes/minikube):
+
+#### Install and start Kubernetes on OSX:
+```bash
+  $ curl -LO https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/darwin/amd64/kubectl && chmod +x kubectl && sudo mv kubectl /usr/local/bin
+  $ curl -Lo minikube https://storage.googleapis.com/minikube/releases/v0.14.0/minikube-darwin-amd64 && chmod +x minikube && sudo mv minikube /usr/local/bin/
+  $ minikube start
+```
+
+#### Or, install and start Kubernetes on Linux:
+```bash
+  $ curl -LO https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/linux/amd64/kubectl && chmod +x kubectl && sudo mv kubectl /usr/local/bin
+  $ curl -Lo minikube https://storage.googleapis.com/minikube/releases/v0.14.0/minikube-linux-amd64 && chmod +x minikube && sudo mv minikube /usr/local/bin/
+  $ minikube start
+```
+
+Or, you can use [Google Container Engine's](https://cloud.google.com/container-engine/) free trial to get a 3 node cluster.
+
+### Verify access to the cluster
+
+```
+  $ kubectl version
+```
+
+### Get and Run Fission: Minikube or Local cluster
+
+If you're using minikube or no cloud provider, use these commands to
+set up services with NodePort.  This exposes fission on ports 31313
+and 31314.
+
+```
+  $ kubectl create -f http://fission.io/fission.yaml
+  $ kubectl create -f http://fission.io/fission-nodeport.yaml
+```
+
+Set the FISSION_URL and FISSION_ROUTER environment variables.
+FISSION_URL is used by the fission CLI to find the server.
+FISSION_URL should be prefixed with a `http://`.  (FISSION_ROUTER is
+only needed for the examples below to work.)
+
+If you're using minikube, use these commands:
+
+```
+  $ export FISSION_URL=http://$(minikube ip):31313
+  $ export FISSION_ROUTER=$(minikube ip):31314
+```
+
+
+### Get and Run Fission: GKE or other Cloud
+
+If you're using GKE or any other cloud provider that supports the
+LoadBalancer service type, use these commands:
+
+```
+  $ kubectl create -f http://fission.io/fission.yaml
+  $ kubectl create -f http://fission.io/fission-cloud.yaml
+```
+
+Save the external IP addresses of controller and router services in
+FISSION_URL and FISSION_ROUTER, respectively.  Wait for services to
+get IP addresses (check this with ```kubectl --namespace fission get
+svc```).  Then:
+
+```
+  $ export FISSION_URL=http://$(kubectl --namespace fission get svc controller -o=jsonpath='{..ip}')
+  $ export FISSION_ROUTER=$(kubectl --namespace fission get svc router -o=jsonpath='{..ip}')
+```
+
+### Install the client CLI
+
+Get the CLI binary for Mac:
+
+```
+  $ curl http://fission.io/mac/fission > fission && chmod +x fission && sudo mv fission /usr/local/bin/
+```
+
+Or Linux:
+
+```
+  $ curl http://fission.io/linux/fission > fission && chmod +x fission && sudo mv fission /usr/local/bin/
+```
+
+### Run an example
+
+Finally, you're ready to use Fission!
+
+```
+  $ fission env create --name nodejs --image fission/node-env
+
+  $ echo 'module.exports = function(context, callback) { callback(200, "Hello, world!\n"); }' > hello.js  
+
+  $ fission function create --name hello --env nodejs --code hello.js
+  
+  $ fission route create --method GET --url /hello --function hello
+  
+  $ curl http://$FISSION_ROUTER/hello
+  Hello, world!
+```

README.md

@@ -1,58 +1,177 @@
-Fission
+Fission: Serverless Functions for Kubernetes
-=======
+============================================
+[![Build Status](https://travis-ci.org/fission/fission.svg?branch=master)](https://travis-ci.org/fission/fission)
+[![Go Report Card](https://goreportcard.com/badge/github.com/fission/fission)](https://goreportcard.com/report/github.com/fission/fission)
+[![Fission Slack](http://slack.fission.io/badge.svg)](http://slack.fission.io)
+
+[fission.io](http://fission.io)  [@fissionio](http://twitter.com/fissionio)
 
 Fission is a fast serverless framework for Kubernetes with a focus on
 developer productivity and high performance.
 
-It's customizable (with sensible defaults), extensible to any
+Fission operates on _just the code_: Docker and Kubernetes are
-language, and interoperates well with other infrastructure.
+abstracted away under normal operation, though you can use both to
+extend Fission if you want to.
 
-See https://fission.io for more.
+Fission is extensible to any language; the core is written in Go, and
+language-specific parts are isolated in something called
+_environments_ (more below).  Fission currently supports NodeJS and
+Python, with more languages coming soon.
+
+### Performance: 100msec cold start
+
+Fission maintains a pool of "warm" containers that each contain a
+small dynamic loader.  When a function is first called,
+i.e. "cold-started", a running container is chosen and the function is
+loaded.  This pool is what makes Fission fast: cold-start latencies
+are typically about 100msec.
+
+### Kubernetes is the right place for Serverless
+
+We're built on Kubernetes because we think any non-trivial app will
+use a combination of serverless functions and more conventional
+microservices, and Kubernetes is a great framework to bring these
+together seamlessly.
+
+Building on Kubernetes also means that anything you do for operations
+on your Kubernetes cluster — such as monitoring or log
+aggregation — also helps with ops on your Fission deployment.
 
 
-Running Fission
+Fission Concepts
-===============
+----------------
 
-On your own Kubernetes cluster
+A _function_ is a piece of code that follows the fission function
-------------------------------
+interface.
+
+An _environment_ contains the language- and runtime-specific parts of
+running a function.  Fission comes with NodeJS and Python
+environments; you can also extend environments or create entirely new
+ones if you want.  (An environment is essentially just a container
+with a webserver and dynamic loader.)
+
+A _trigger_ is something that maps an event to a function; Fission
+supports HTTP routes as triggers today, with upcoming support for
+other types of event triggers, such as timers and Kubernetes events.
+
+Usage
+-----
+
+```bash
+
+  # Add the stock NodeJS env to your Fission deployment
+  $ fission env create --name nodejs --image fission/node-env
+
+  # A javascript one-liner that prints "hello world"
+  $ echo 'module.exports = function(context, callback) { callback(200, "Hello, world!\n"); }' > hello.js  
+
+  # Upload your function code to fission
+  $ fission function create --name hello --env nodejs --code hello.js
+
+  # Map GET /hello to your new function
+  $ fission route create --method GET --url /hello --function hello
+
+  # Run the function.  This takes about 100msec the first time.
+  $ curl http://$FISSION_ROUTER/hello
+  Hello, world!
+```
+
+See the [examples](examples) directory for more.
+
+Running Fission on your Cluster
+===============================
 
 ### Setup Kubernetes
 
-On your laptop:
+You can install Kubernetes on your laptop with [minikube](https://github.com/kubernetes/minikube):
 
-  https://github.com/kubernetes/minikube
+#### Install and start Kubernetes on OSX:
+```bash
+  $ curl -LO https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/darwin/amd64/kubectl && chmod +x kubectl && sudo mv kubectl /usr/local/bin
+  $ curl -Lo minikube https://storage.googleapis.com/minikube/releases/v0.14.0/minikube-darwin-amd64 && chmod +x minikube && sudo mv minikube /usr/local/bin/
+  $ minikube start
+```
 
-Or use Google Container Engine.
+#### Or, install and start Kubernetes on Linux:
+```bash
+  $ curl -LO https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/linux/amd64/kubectl && chmod +x kubectl && sudo mv kubectl /usr/local/bin
+  $ curl -Lo minikube https://storage.googleapis.com/minikube/releases/v0.14.0/minikube-linux-amd64 && chmod +x minikube && sudo mv minikube /usr/local/bin/
+  $ minikube start
+```
+
+Or, you can use [Google Container Engine's](https://cloud.google.com/container-engine/) free trial to get a 3 node cluster.
 
 ### Verify access to the cluster
 
+```
   $ kubectl version
+```
 
-### Get and Run Fission
+### Get and Run Fission: Minikube or Local cluster
 
-If you're using GKE, use the fission.yaml unmodified.  If you're using
+If you're using minikube or no cloud provider, use these commands to
-minikube, change all instances of LoadBalancer services to NodePort.
+set up services with NodePort.  This exposes fission on ports 31313
+and 31314.
+
+```
+  $ kubectl create -f http://fission.io/fission.yaml
+  $ kubectl create -f http://fission.io/fission-nodeport.yaml
+```
+
+Set the FISSION_URL and FISSION_ROUTER environment variables.
+FISSION_URL is used by the fission CLI to find the server.
+FISSION_URL should be prefixed with a `http://`.  (FISSION_ROUTER is
+only needed for the examples below to work.)
+
+If you're using minikube, use these commands:
+
+```
+  $ export FISSION_URL=http://$(minikube ip):31313
+  $ export FISSION_ROUTER=$(minikube ip):31314
+```
 
 
-  $ curl http://fission.io/fission.yaml | kubectl create -f -
+### Get and Run Fission: GKE or other Cloud
 
-  $ kubectl --namespace fission get services
+If you're using GKE or any other cloud provider that supports the
+LoadBalancer service type, use these commands:
+
+```
+  $ kubectl create -f http://fission.io/fission.yaml
+  $ kubectl create -f http://fission.io/fission-cloud.yaml
+```
 
 Save the external IP addresses of controller and router services in
-FISSION_URL and FISSION_ROUTER, respectively.
+FISSION_URL and FISSION_ROUTER, respectively.  Wait for services to
+get IP addresses (check this with ```kubectl --namespace fission get
+svc```).  Then:
+
+```
+  $ export FISSION_URL=http://$(kubectl --namespace fission get svc controller -o=jsonpath='{..ip}')
+  $ export FISSION_ROUTER=$(kubectl --namespace fission get svc router -o=jsonpath='{..ip}')
+```
 
 ### Install the client CLI
 
-  $ curl http://fission.io/fission > fission
+Get the CLI binary for Mac:
 
-  $ chmod +x fission
+```
+  $ curl http://fission.io/mac/fission > fission && chmod +x fission && sudo mv fission /usr/local/bin/
+```
 
-  $ sudo mv fission /usr/local/bin/
+Or Linux:
+
+```
+  $ curl http://fission.io/linux/fission > fission && chmod +x fission && sudo mv fission /usr/local/bin/
+```
 
 ### Run an example
 
+Finally, you're ready to use Fission!
+
+```
   $ fission env create --name nodejs --image fission/node-env
-  
+
   $ echo 'module.exports = function(context, callback) { callback(200, "Hello, world!\n"); }' > hello.js  
 
   $ fission function create --name hello --env nodejs --code hello.js
@@ -61,19 +180,59 @@ FISSION_URL and FISSION_ROUTER, respectively.
   
   $ curl http://$FISSION_ROUTER/hello
   Hello, world!
+```
 
+Compiling Fission
+=================
+
+[You only need to do this if you're making Fission changes; if you're
+just deploying Fission, use fission.yaml which points to prebuilt
+images.]
+
+You'll need go installed, along with the glide dependecy management
+tool.  You'll also need docker for building images.
+
+The server side is compiled as one binary ("fission-bundle") which
+contains controller, poolmgr and router; it invokes the right one
+based on command-line arguments.
+
+To build fission-bundle: clone this repo, then from the top level
+directory:
+
+```
+  # Get dependencies
+  $ glide install
+
+  # Build fission server and an image
+  $ pushd fission-bundle
+  $ ./build.sh
+
+  # Edit push.sh to point to your registry
+  $ $EDITOR push.sh
+  $ ./push.sh
+  $ popd
+
+  # To install, update fission.yaml to point to your compiled image
+  $ $EDITOR fission.yaml
+  $ kubectl create -f fission.yaml
+```
+
+If you're changing the CLI:
+
+```
+  # Build Fission CLI
+  $ cd fission && go install
+```
 
 Status
 ======
 
-Fission is in early alpha.  Watch this space for announcements soon!
+Fission is in early alpha.  It's not suitable for production use just
+yet.  
 
-
+We're looking for early developer feedback -- if you do use Fission,
-Performance
+we'd love to hear how it's working for you, what parts in particular
-===========
+you'd like to see improved, and so on.  Talk to us on
-
+[slack](http://slack.fission.io) or
-The alpha release has focussed on cold-start latency performance.  For
+[twitter](https://twitter.com/fissionio).
-requests that don't have a running instance, i.e. a "cold start",
-fission has a latency overhead of less than 100 msec for the NodeJS
-environment.
 

cache/cache.go

@@ -21,7 +21,7 @@ import (
 
 	"errors"
 	"fmt"
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 type requestType int

cache/cache_test.go

@@ -27,11 +27,11 @@ func checkErr(err error) {
 }
 
 func TestCache(t *testing.T) {
-	c := MakeCache(100 * time.Millisecond)
+	c := MakeCache(100*time.Millisecond, 100*time.Millisecond)
 
-	err := c.Set("a", "b")
+	err, _ := c.Set("a", "b")
 	checkErr(err)
-	err = c.Set("p", "q")
+	err, _ = c.Set("p", "q")
 	checkErr(err)
 
 	val, err := c.Get("a")
@@ -53,7 +53,7 @@ func TestCache(t *testing.T) {
 		log.Panicf("found deleted element")
 	}
 
-	err = c.Set("expires", "42")
+	err, _ = c.Set("expires", "42")
 	checkErr(err)
 	time.Sleep(150 * time.Millisecond)
 	_, err = c.Get("expires")

common.go

@@ -0,0 +1,30 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package fission
+
+import (
+	"fmt"
+)
+
+func UrlForFunction(m *Metadata) string {
+	prefix := "/fission-function"
+	if len(m.Uid) > 0 {
+		return fmt.Sprintf("%v/%v/%v", prefix, m.Name, m.Uid)
+	} else {
+		return fmt.Sprintf("%v/%v", prefix, m.Name)
+	}
+}

controller/api.go

@@ -26,13 +26,24 @@ import (
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 type API struct {
 	FunctionStore
 	HTTPTriggerStore
 	EnvironmentStore
+	WatchStore
+}
+
+func MakeAPI(rs *ResourceStore) *API {
+	api := &API{
+		FunctionStore:    FunctionStore{ResourceStore: *rs},
+		HTTPTriggerStore: HTTPTriggerStore{ResourceStore: *rs},
+		EnvironmentStore: EnvironmentStore{ResourceStore: *rs},
+		WatchStore:       WatchStore{ResourceStore: *rs},
+	}
+	return api
 }
 
 func (api *API) respondWithSuccess(w http.ResponseWriter, resp []byte) {
@@ -51,7 +62,7 @@ func (api *API) respondWithError(w http.ResponseWriter, err error) {
 }
 
 func (api *API) HomeHandler(w http.ResponseWriter, r *http.Request) {
-	fmt.Fprintf(w, "{message: \"Fission API\", version: \"0.1.0\"}\n")
+	fmt.Fprintf(w, "{\"message\": \"Fission API\", \"version\": \"0.1.0\"}\n")
 }
 
 func (api *API) Serve(port int) {
@@ -76,6 +87,12 @@ func (api *API) Serve(port int) {
 	r.HandleFunc("/v1/environments/{environment}", api.EnvironmentApiUpdate).Methods("PUT")
 	r.HandleFunc("/v1/environments/{environment}", api.EnvironmentApiDelete).Methods("DELETE")
 
+	r.HandleFunc("/v1/watches", api.WatchApiList).Methods("GET")
+	r.HandleFunc("/v1/watches", api.WatchApiCreate).Methods("POST")
+	r.HandleFunc("/v1/watches/{watch}", api.WatchApiGet).Methods("GET")
+	r.HandleFunc("/v1/watches/{watch}", api.WatchApiUpdate).Methods("PUT")
+	r.HandleFunc("/v1/watches/{watch}", api.WatchApiDelete).Methods("DELETE")
+
 	address := fmt.Sprintf(":%v", port)
 
 	log.WithFields(log.Fields{"port": port}).Info("Server started")

controller/api_test.go

@@ -28,8 +28,8 @@ import (
 	etcdClient "github.com/coreos/etcd/client"
 	"golang.org/x/net/context"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
-	"github.com/platform9/fission/controller/client"
+	"github.com/fission/fission/controller/client"
 )
 
 var g struct {
@@ -185,22 +185,55 @@ func TestEnvironmentApi(t *testing.T) {
 	assert(len(ts) == 2, "created two envs, but didn't find them")
 }
 
+func TestWatchApi(t *testing.T) {
+	testWatch := &fission.Watch{
+		Metadata: fission.Metadata{
+			Name: "xxx",
+			Uid:  "yyy",
+		},
+		Namespace:     "default",
+		ObjType:       "pod",
+		LabelSelector: "",
+		FieldSelector: "",
+		Function: fission.Metadata{
+			Name: "foo",
+			Uid:  "",
+		},
+		Target: "",
+	}
+	m, err := g.client.WatchCreate(testWatch)
+	panicIf(err)
+	defer g.client.WatchDelete(m)
+
+	w, err := g.client.WatchGet(m)
+	panicIf(err)
+	testWatch.Metadata.Uid = m.Uid
+	w.Target = ""
+	assert(*testWatch == *w, "watch should match after reading")
+
+	testWatch.Metadata.Name = "yyy"
+	m2, err := g.client.WatchCreate(testWatch)
+	panicIf(err)
+	defer g.client.WatchDelete(m2)
+
+	ws, err := g.client.WatchList()
+	panicIf(err)
+	assert(len(ws) == 2, "created two envs, but didn't find them")
+}
+
 func TestMain(m *testing.M) {
 	flag.Parse()
 
 	fileStore, ks, rs := getTestResourceStore()
 	defer os.RemoveAll(fileStore.root)
 
-	api := &API{
+	api := MakeAPI(rs)
-		FunctionStore:    FunctionStore{ResourceStore: *rs},
-		HTTPTriggerStore: HTTPTriggerStore{ResourceStore: *rs},
-		EnvironmentStore: EnvironmentStore{ResourceStore: *rs},
-	}
 	g.client = client.MakeClient("http://localhost:8888")
 
 	ks.Delete(context.Background(), "Function", &etcdClient.DeleteOptions{Recursive: true})
 	ks.Delete(context.Background(), "HTTPTrigger", &etcdClient.DeleteOptions{Recursive: true})
 	ks.Delete(context.Background(), "Environment", &etcdClient.DeleteOptions{Recursive: true})
+	ks.Delete(context.Background(), "Watch", &etcdClient.DeleteOptions{Recursive: true})
 
 	go api.Serve(8888)
 	time.Sleep(500 * time.Millisecond)

controller/client/client.go

@@ -28,7 +28,7 @@ import (
 
 	log "github.com/Sirupsen/logrus"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 type (
@@ -459,3 +459,90 @@ func (c *Client) EnvironmentList() ([]fission.Environment, error) {
 
 	return envs, nil
 }
+
+func (c *Client) WatchCreate(w *fission.Watch) (*fission.Metadata, error) {
+	reqbody, err := json.Marshal(w)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := http.Post(c.url("watches"), "application/json", bytes.NewReader(reqbody))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	var m fission.Metadata
+	err = json.Unmarshal(body, &m)
+	if err != nil {
+		return nil, err
+	}
+
+	return &m, nil
+}
+
+func (c *Client) WatchGet(m *fission.Metadata) (*fission.Watch, error) {
+	relativeUrl := fmt.Sprintf("watches/%v", m.Name)
+	if len(m.Uid) > 0 {
+		relativeUrl += fmt.Sprintf("?uid=%v", m.Uid)
+	}
+
+	resp, err := http.Get(c.url(relativeUrl))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	var w fission.Watch
+	err = json.Unmarshal(body, &w)
+	if err != nil {
+		return nil, err
+	}
+
+	return &w, nil
+}
+
+func (c *Client) WatchUpdate(w *fission.Watch) (*fission.Metadata, error) {
+	return nil, fission.MakeError(fission.ErrorNotImplmented,
+		"watch update not implemented")
+}
+
+func (c *Client) WatchDelete(m *fission.Metadata) error {
+	relativeUrl := fmt.Sprintf("watches/%v", m.Name)
+	if len(m.Uid) > 0 {
+		relativeUrl += fmt.Sprintf("?uid=%v", m.Uid)
+	}
+	err := c.delete(relativeUrl)
+	return err
+
+}
+
+func (c *Client) WatchList() ([]fission.Watch, error) {
+	resp, err := http.Get(c.url("watches"))
+	if err != nil {
+		return nil, err
+	}
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	watches := make([]fission.Watch, 0)
+	err = json.Unmarshal(body, &watches)
+	if err != nil {
+		return nil, err
+	}
+
+	return watches, err
+}

controller/environmentApi.go

@@ -24,7 +24,7 @@ import (
 	log "github.com/Sirupsen/logrus"
 	"github.com/gorilla/mux"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 func (api *API) EnvironmentApiList(w http.ResponseWriter, r *http.Request) {

controller/environmentStore.go

@@ -19,7 +19,7 @@ package controller
 import (
 	"github.com/satori/go.uuid"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 type EnvironmentStore struct {

controller/functionApi.go

@@ -25,7 +25,7 @@ import (
 	"github.com/gorilla/mux"
 
 	"encoding/base64"
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 func (api *API) FunctionApiList(w http.ResponseWriter, r *http.Request) {

controller/functionStore.go

@@ -19,7 +19,7 @@ package controller
 import (
 	log "github.com/Sirupsen/logrus"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 type FunctionStore struct {

controller/httpTriggerApi.go

@@ -24,7 +24,7 @@ import (
 	log "github.com/Sirupsen/logrus"
 	"github.com/gorilla/mux"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 func (api *API) HTTPTriggerApiList(w http.ResponseWriter, r *http.Request) {

controller/httpTriggerStore.go

@@ -19,7 +19,7 @@ package controller
 import (
 	"github.com/satori/go.uuid"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 type HTTPTriggerStore struct {

controller/watchApi.go

@@ -0,0 +1,121 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/gorilla/mux"
+
+	"github.com/fission/fission"
+)
+
+func (api *API) WatchApiList(w http.ResponseWriter, r *http.Request) {
+	watches, err := api.WatchStore.List()
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	resp, err := json.Marshal(watches)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	api.respondWithSuccess(w, resp)
+}
+
+func (api *API) WatchApiCreate(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		api.respondWithError(w, err)
+	}
+
+	var watch fission.Watch
+	err = json.Unmarshal(body, &watch)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	watch.Target = fission.UrlForFunction(&watch.Function)
+
+	uid, err := api.WatchStore.Create(&watch)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	m := &fission.Metadata{Name: watch.Metadata.Name, Uid: uid}
+	resp, err := json.Marshal(m)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	api.respondWithSuccess(w, resp)
+}
+
+func (api *API) WatchApiGet(w http.ResponseWriter, r *http.Request) {
+	var m fission.Metadata
+
+	vars := mux.Vars(r)
+	m.Name = vars["watch"]
+	m.Uid = r.FormValue("uid") // empty if uid is absent
+
+	watch, err := api.WatchStore.Get(&m)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	resp, err := json.Marshal(watch)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	api.respondWithSuccess(w, resp)
+}
+
+func (api *API) WatchApiUpdate(w http.ResponseWriter, r *http.Request) {
+	api.respondWithError(w, fission.MakeError(fission.ErrorNotImplmented,
+		"Not implemented"))
+}
+
+func (api *API) WatchApiDelete(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	var m fission.Metadata
+	m.Name = vars["watch"]
+
+	m.Uid = r.FormValue("uid") // empty if uid is absent
+	if len(m.Uid) == 0 {
+		log.WithFields(log.Fields{"watch": m.Name}).Info("Deleting all versions")
+	}
+
+	err := api.WatchStore.Delete(m)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	api.respondWithSuccess(w, []byte(""))
+}

controller/watchStore.go

@@ -0,0 +1,79 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"github.com/satori/go.uuid"
+
+	"github.com/fission/fission"
+)
+
+type WatchStore struct {
+	ResourceStore
+}
+
+func (ws *WatchStore) Create(w *fission.Watch) (string, error) {
+	w.Metadata.Uid = uuid.NewV4().String()
+	return w.Metadata.Uid, ws.ResourceStore.create(w)
+}
+
+func (ws *WatchStore) Get(m *fission.Metadata) (*fission.Watch, error) {
+	var w fission.Watch
+	err := ws.ResourceStore.read(m.Name, &w)
+	if err != nil {
+		return nil, err
+	}
+	return &w, err
+}
+
+func (ws *WatchStore) Update(w *fission.Watch) (string, error) {
+	w.Metadata.Uid = uuid.NewV4().String()
+	return w.Metadata.Uid, ws.ResourceStore.update(w)
+}
+
+func (ws *WatchStore) Delete(m fission.Metadata) error {
+	typeName, err := getTypeName(fission.Watch{})
+	if err != nil {
+		return err
+	}
+	return ws.ResourceStore.delete(typeName, m.Name)
+}
+
+func (ws *WatchStore) List() ([]fission.Watch, error) {
+	typeName, err := getTypeName(fission.Watch{})
+	if err != nil {
+		return nil, err
+	}
+
+	bufs, err := ws.ResourceStore.getAll(typeName)
+	if err != nil {
+		return nil, err
+	}
+
+	watches := make([]fission.Watch, 0, len(bufs))
+	js := JsonSerializer{}
+	for _, buf := range bufs {
+		var w fission.Watch
+		err = js.deserialize([]byte(buf), &w)
+		if err != nil {
+			return nil, err
+		}
+		watches = append(watches, w)
+	}
+
+	return watches, nil
+}

environments/nodejs/server.js

@@ -82,6 +82,7 @@ app.all('/', function (req, res) {
     try {
         userFunction(context, callback);
     } catch(e) {
+        console.log(`Function error: ${e}`);
         callback(500, "Internal server error")
     }
 });

environments/python3/Dockerfile

@@ -1,3 +1,13 @@
-FROM python:3-onbuild
+FROM ubuntu:16.04
-CMD [ "python", "./server.py" ]
+
-EXPOSE 8888
+RUN apt-get update -y
+RUN apt-get install -y python3 python3-pip python3-dev build-essential
+RUN pip3 install --upgrade pip
+
+COPY . /app
+WORKDIR /app
+RUN pip3 install -r requirements.txt
+
+ENTRYPOINT ["python3"]
+CMD ["server.py"]
+

environments/python3/README.md

@@ -0,0 +1,45 @@
+# Fission: Python Environment
+
+This is the Python environment for Fission.
+
+It's a Docker image containing a Python 3.5 runtime, along with a
+dynamic loader.  A few common dependencies are included in the
+requirements.txt file.
+
+## Customizing this image
+
+To add package dependencies, edit requirements.txt to add what you
+need, and rebuild this image (instructions below).
+
+You also may want to customize what's available to the function in its
+request context.  You can do this by editing server.py (see the
+comment in that file about customizing request context).
+
+## Rebuilding and pushing the image
+
+You'll need access to a Docker registry to push the image: you can
+sign up for Docker hub at hub.docker.com, or use registries from
+gcr.io, quay.io, etc.  Let's assume you're using a docker hub account
+called USER.  Build and push the image to the the registry:
+
+```
+   docker build -t USER/python-env . && docker push USER/python-env
+```
+
+## Using the image in fission
+
+You can add this customized image to fission with "fission env
+create":
+
+```
+   fission env create --name python --image USER/python-env
+```
+
+Or, if you already have an environment, you can update its image:
+
+```
+   fission env update --name python --image USER/python-env   
+```
+
+After this, fission functions that have the env parmeter set to the
+same environment name as this command will use this environment.

environments/python3/requirements.txt

@@ -0,0 +1,6 @@
+Flask===0.11.1
+httplib2
+python-dateutil
+requests==2.7.0
+redis
+hiredis

environments/python3/server.py

@@ -1,42 +1,55 @@
 #!/usr/bin/env python
 
+import logging
+import sys
 import imp
-from http.server import BaseHTTPRequestHandler, HTTPServer
 
-#
+from flask import Flask, request, abort, g
-# Load the file.
+
-#
+app = Flask(__name__)
+
 codepath = '/userfunc/user'
+
 userfunc = None
 
-class testHTTPServer_RequestHandler(BaseHTTPRequestHandler):
+@app.route('/specialize', methods=['POST'])
-    def do_POST(self):
+def load():
-        global userfunc
+    global userfunc
-        userfunc = (imp.load_source('user', codepath)).main
+    userfunc = (imp.load_source('user', codepath)).main
-        self.send_response(200)
+    return ""
-        self.end_headers()
-        self.wfile.write(bytes("ok\n", "utf8"))
-        return
 
-    # GET
+@app.route('/', methods=['GET', 'POST', 'PUT', 'HEAD', 'OPTIONS', 'DELETE'])
-    def do_GET(self):
+def f():
-        global userfunc
+    if userfunc == None:
-        try:
+        print("Generic container: no requests supported")
-            print("GET request")
+        abort(500)
-            message = userfunc(None)
+    #
-            self.send_response(200)
+    # Customizing the request context
-            self.end_headers()
+    #
-            self.wfile.write(bytes(message, "utf8"))
+    # If you want to pass something to the function, you can add it to 'g':
-            return
+    #   g.myKey = myValue
-        except:
+    # And the user func can then access that (after doing a "from flask import g").
-            self.send_response(500)
+    #
-            self.end_headers()
+    return userfunc()
 
-def run():
+#
-    print('starting server...')
+# Logging setup.  TODO: Loglevel hard-coded for now. We could allow
-    server_address = ('0.0.0.0', 8888)
+# functions/routes to override this somehow; or we could create
-    httpd = HTTPServer(server_address, testHTTPServer_RequestHandler)
+# separate dev vs. prod environments.
-    httpd.serve_forever()
+#
+def setup_logger(loglevel):
+    global app
+    root = logging.getLogger()
+    root.setLevel(loglevel)
+    ch = logging.StreamHandler(sys.stdout)
+    ch.setLevel(loglevel)
+    ch.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
+    app.logger.addHandler(ch)
 
-
+#
-run()
+# TODO: this starts the built-in server, which isn't the most
+# efficient.  We should use something better.
+#
+setup_logger(logging.DEBUG)
+app.logger.info("Starting server")
+app.run(host='0.0.0.0', port='8888')

examples/python/guestbook/add.py

@@ -0,0 +1,16 @@
+#
+# Handles POST /guestbook -- adds item to guestbook 
+#
+
+from flask import request, redirect
+import redis
+
+# Connect to redis.
+redisConnection = redis.StrictRedis(host='redis.guestbook', port=6379, db=0)
+
+def main():
+    # Read the item from POST params, add it to redis, and redirect
+    # back to the list
+    item = request.form['text']
+    redisConnection.rpush('guestbook', item)
+    return redirect('/guestbook', code=303)

examples/python/guestbook/deploy.sh

@@ -0,0 +1,18 @@
+#!/bin/sh
+
+set -e
+
+kubectl create -f redis.yaml
+
+if [ -z "$FISSION_URL" ]
+then
+    echo "Need $FISSION_URL set to a fission controller address"
+    exit 1
+fi
+
+# Create python env if it doesn't exist
+fission env get --name python || fission env create --name python --image fission/python-env
+
+# Register functions and routes with fission
+fission function create --name guestbook-get --env python --code get.py --url /guestbook --method GET
+fission function create --name guestbook-add --env python --code add.py --url /guestbook --method POST

examples/python/guestbook/get.py

@@ -0,0 +1,28 @@
+#
+# Handles GET /guestbook -- returns a list of items in the guestbook
+# with a form to add more.
+#
+
+from flask import current_app, escape
+import redis
+
+# Connect to redis.  This is run only when this file is loaded; as
+# long as the pod is alive, the connection is reused.
+redisConnection = redis.StrictRedis(host='redis.guestbook', port=6379, db=0)
+
+def main():
+    messages = redisConnection.lrange('guestbook', 0, -1)
+
+    items = [("<li>%s</li>" % escape(m.decode('utf-8'))) for m in messages]
+    ul = "<ul>%s</ul>" % "\n".join(items)
+    return """
+      <html><body>
+          <h1>Guestbook</h1>      
+          <form action="/guestbook" method="POST">
+            <input type="text" name="text">
+            <button type="submit">Add</button>
+          </form>
+          <hr/>
+          %s
+      </body></html>
+      """ % ul

examples/python/guestbook/redis.yaml

@@ -0,0 +1,45 @@
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: guestbook
+  labels:
+    name: guestbook
+
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  labels:
+    run: redis
+  name: redis
+  namespace: guestbook
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      run: redis
+  template:
+    metadata:
+      labels:
+        run: redis
+    spec:
+      containers:
+      - image: redis
+        name: redis
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    run: redis
+  name: redis
+  namespace: guestbook
+spec:
+  selector:
+    run: redis
+  type: ClusterIP
+  ports:
+  - port: 6379
+    protocol: TCP
+    targetPort: 6379

examples/python/hello.py

@@ -1,2 +1,2 @@
-def main(context):
+def main():
-    return "Hello, world!"
+    return "Hello, world!\n"

examples/python/requestdata.py

@@ -0,0 +1,7 @@
+from flask import request
+from flask import current_app
+
+def main():
+    current_app.logger.info("Received request")
+    msg = "---HEADERS---\n%s\n--BODY--\n%s\n-----\n" % (request.headers, request.get_data())
+    return msg

fission-bundle/main.go

@@ -2,25 +2,17 @@ package main
 
 import (
 	"log"
-	"os"
 	"strconv"
-	//"time"
 
 	"github.com/docopt/docopt-go"
-	"github.com/platform9/fission/controller"
+	"github.com/fission/fission/controller"
-	"github.com/platform9/fission/poolmgr"
+	"github.com/fission/fission/kubewatcher"
-	"github.com/platform9/fission/router"
+	"github.com/fission/fission/poolmgr"
+	"github.com/fission/fission/router"
 )
 
 func runController(port int, etcdUrl string, filepath string) {
-	_, err := os.Stat(filepath)
+	// filePath will be created if it doesn't exist.
-	if err != nil {
-		if os.IsNotExist(err) {
-			log.Fatalf("Error: path %v does not exist", filepath)
-		} else {
-			log.Fatalf("Error: can't access path %v", filepath)
-		}
-	}
 	fileStore := controller.MakeFileStore(filepath)
 
 	rs, err := controller.MakeResourceStore(fileStore, []string{etcdUrl})
@@ -28,11 +20,7 @@ func runController(port int, etcdUrl string, filepath string) {
 		log.Fatalf("Error: %v", err)
 	}
 
-	api := &controller.API{
+	api := controller.MakeAPI(rs)
-		FunctionStore:    controller.FunctionStore{ResourceStore: *rs},
-		HTTPTriggerStore: controller.HTTPTriggerStore{ResourceStore: *rs},
-		EnvironmentStore: controller.EnvironmentStore{ResourceStore: *rs},
-	}
 	api.Serve(port)
 	log.Fatalf("Error: Controller exited.")
 }
@@ -49,6 +37,13 @@ func runPoolmgr(port int, controllerUrl string, namespace string) {
 	}
 }
 
+func runKubeWatcher(controllerUrl, routerUrl string) {
+	err := kubewatcher.Start(controllerUrl, routerUrl)
+	if err != nil {
+		log.Fatalf("Error starting kubewatcher: %v", err)
+	}
+}
+
 func getPort(portArg interface{}) int {
 	portArgStr := portArg.(string)
 	port, err := strconv.Atoi(portArgStr)
@@ -80,16 +75,19 @@ Use it to start one or more of the fission servers:
 Usage:
   fission-bundle --controllerPort=<port> [--etcdUrl=<etcdUrl>] --filepath=<filepath>
   fission-bundle --routerPort=<port> [--controllerUrl=<url> --poolmgrUrl=<url>]
-  fission-bundle --poolmgrPort=<port> [--controllerUrl=<url>]
+  fission-bundle --poolmgrPort=<port> [--controllerUrl=<url> --namespace=<namespace>]
+  fission-bundle --kubewatcher [--controllerUrl=<url> --routerUrl=<url>]
 Options:
   --controllerPort=<port>  Port that the controller should listen on.
   --routerPort=<port>      Port that the router should listen on.
   --poolmgrPort=<port>     Port that the poolmgr should listen on.
   --controllerUrl=<url>    Controller URL. Not required if --controllerPort is specified.
-  --poolmgrUrl=<url>       Controller URL. Not required if --poolmgrPort is specified.
+  --poolmgrUrl=<url>       Poolmgr URL. Not required if --poolmgrPort is specified.
+  --routerUrl=<url>        Router URL.
   --etcdUrl=<etcdUrl>      Etcd URL.
   --filepath=<filepath>    Directory to store functions in.
   --namespace=<namespace>  Kubernetes namespace in which to run function containers. Defaults to 'fission-function'.
+  --kubewatcher            Start Kubernetes events watcher.
 `
 	arguments, err := docopt.Parse(usage, nil, true, "fission-bundle", false)
 	if err != nil {
@@ -101,6 +99,7 @@ Options:
 	controllerUrl := getStringArgWithDefault(arguments["--controllerUrl"], "http://controller.fission")
 	etcdUrl := getStringArgWithDefault(arguments["--etcdUrl"], "http://etcd:2379")
 	poolmgrUrl := getStringArgWithDefault(arguments["--poolmgrUrl"], "http://poolmgr.fission")
+	routerUrl := getStringArgWithDefault(arguments["--routerUrl"], "http://router.fission")
 
 	if arguments["--controllerPort"] != nil {
 		port := getPort(arguments["--controllerPort"])
@@ -117,5 +116,9 @@ Options:
 		runPoolmgr(port, controllerUrl, namespace)
 	}
 
+	if arguments["--kubewatcher"] == true {
+		runKubeWatcher(controllerUrl, routerUrl)
+	}
+
 	select {}
 }

fission-bundle/push.sh

@@ -2,8 +2,14 @@
 
 set -e
 
+tag=$1
+if [ -z "$tag" ]
+then
+    tag=latest
+fi
+
 . build.sh
 
 docker build -t fission-bundle .
-docker tag fission-bundle fission/fission-bundle
+docker tag fission-bundle fission/fission-bundle:$tag
-docker push fission/fission-bundle
+docker push fission/fission-bundle:$tag

fission-cloud.yaml

@@ -0,0 +1,32 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: router
+  namespace: fission
+  labels:
+    svc: router
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 80
+    targetPort: 8888
+    nodePort: 31314
+  selector:
+    svc: router
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: controller
+  namespace: fission
+  labels:
+    svc: controller
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 80
+    targetPort: 8888
+    nodePort: 31313
+  selector:
+    svc: controller

fission-nodeport.yaml

@@ -0,0 +1,32 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: router
+  namespace: fission
+  labels:
+    svc: router
+spec:
+  type: NodePort
+  ports:
+  - port: 80
+    targetPort: 8888
+    nodePort: 31314
+  selector:
+    svc: router
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: controller
+  namespace: fission
+  labels:
+    svc: controller
+spec:
+  type: NodePort
+  ports:
+  - port: 80
+    targetPort: 8888
+    nodePort: 31313
+  selector:
+    svc: controller

fission.yaml

@@ -13,23 +13,6 @@ metadata:
   labels:
     name: fission-function
 
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: controller
-  namespace: fission
-  labels:
-    svc: controller
-spec:
-  type: LoadBalancer
-  ports:
-  - port: 80
-    targetPort: 8888
-    nodePort: 31313
-  selector:
-    svc: controller
-
 ---
 apiVersion: extensions/v1beta1
 kind: Deployment
@@ -39,33 +22,16 @@ metadata:
 spec:
   replicas: 1
   template:
-    metadata: 
+    metadata:
       labels:
         svc: controller
     spec:
       containers:
       - name: controller
-        image: fission/fission-bundle
+        image: fission/fission-bundle:alpha10
         command: ["/fission-bundle"]
         args: ["--controllerPort", "8888", "--filepath", "/filestore"]
 
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: router
-  namespace: fission
-  labels:
-    svc: router
-spec:
-  type: LoadBalancer
-  ports:
-  - port: 80
-    targetPort: 8888
-    nodePort: 31314
-  selector:
-    svc: router
-
 ---
 apiVersion: extensions/v1beta1
 kind: Deployment
@@ -75,13 +41,13 @@ metadata:
 spec:
   replicas: 1
   template:
-    metadata: 
+    metadata:
       labels:
         svc: router
     spec:
       containers:
       - name: router
-        image: fission/fission-bundle
+        image: fission/fission-bundle:alpha10
         command: ["/fission-bundle"]
         args: ["--routerPort", "8888"]
 
@@ -109,16 +75,35 @@ metadata:
 spec:
   replicas: 1
   template:
-    metadata: 
+    metadata:
       labels:
         svc: poolmgr
     spec:
       containers:
       - name: poolmgr
-        image: fission/fission-bundle
+        image: fission/fission-bundle:alpha10
         command: ["/fission-bundle"]
         args: ["--poolmgrPort", "8888"]
 
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: kubewatcher
+  namespace: fission
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        svc: kubewatcher
+    spec:
+      containers:
+      - name: kubewatcher
+        image: fission/fission-bundle:alpha10
+        command: ["/fission-bundle"]
+        args: ["--kubewatcher"]
+
 ---
 apiVersion: v1
 kind: Service

fission/common.go

@@ -19,8 +19,9 @@ package main
 import (
 	"fmt"
 	"os"
+	"strings"
 
-	"github.com/platform9/fission/controller/client"
+	"github.com/fission/fission/controller/client"
 )
 
 func fatal(msg string) {
@@ -34,6 +35,8 @@ func getClient(serverUrl string) *client.Client {
 		fatal("Need --server or FISSION_URL set to your fission server.")
 	}
 
+	serverUrl = "http://" + strings.TrimPrefix(serverUrl, "http://")
+
 	return client.MakeClient(serverUrl)
 }
 

fission/environment.go

@@ -22,7 +22,7 @@ import (
 
 	"github.com/urfave/cli"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 	"os"
 )
 

fission/function.go

@@ -20,11 +20,13 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"os/exec"
 	"text/tabwriter"
 
+	"github.com/satori/go.uuid"
 	"github.com/urfave/cli"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 func fnCreate(c *cli.Context) error {
@@ -58,6 +60,31 @@ func fnCreate(c *cli.Context) error {
 	checkErr(err, "create function")
 
 	fmt.Printf("function '%v' created\n", fnName)
+
+	// Allow the user to specify an HTTP trigger while creating a function.
+	triggerUrl := c.String("url")
+	if len(triggerUrl) == 0 {
+		return nil
+	}
+	method := c.String("method")
+	if len(method) == 0 {
+		method = "GET"
+	}
+	triggerName := uuid.NewV4().String()
+	ht := &fission.HTTPTrigger{
+		Metadata: fission.Metadata{
+			Name: triggerName,
+		},
+		UrlPattern: triggerUrl,
+		Method:     getMethod(method),
+		Function: fission.Metadata{
+			Name: fnName,
+		},
+	}
+	_, err = client.HTTPTriggerCreate(ht)
+	checkErr(err, "create HTTP trigger")
+	fmt.Printf("route created: %v %v -> %v\n", method, triggerUrl, fnName)
+
 	return err
 }
 
@@ -128,6 +155,10 @@ func fnDelete(c *cli.Context) error {
 	client := getClient(c.GlobalString("server"))
 
 	fnName := c.String("name")
+	if len(fnName) == 0 {
+		fatal("Need name of function, use --name")
+	}
+
 	fnUid := c.String("uid")
 	m := &fission.Metadata{Name: fnName, Uid: fnUid}
 
@@ -155,3 +186,55 @@ func fnList(c *cli.Context) error {
 
 	return err
 }
+
+func fnEdit(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+
+	fnName := c.String("name")
+	if len(fnName) == 0 {
+		fatal("Need name of function, use --name")
+	}
+	fnUid := c.String("uid")
+
+	// get function meta
+	function, err := client.FunctionGet(&fission.Metadata{Name: fnName, Uid: fnUid})
+	checkErr(err, fmt.Sprintf("read function '%v'", fnName))
+
+	// write to tmp file
+	tmpFile, err := ioutil.TempFile("", fnName)
+	checkErr(err, "create temp file")
+	defer os.Remove(tmpFile.Name())
+
+	_, err = tmpFile.Write([]byte(function.Code))
+	checkErr(err, "write temp file")
+	tmpFile.Close()
+
+	// invoke $EDITOR on tmp file and wait for it
+	editor := os.Getenv("EDITOR")
+	if len(editor) == 0 {
+		editor = "vi"
+	}
+	cmd := exec.Command("/bin/sh", "-c", fmt.Sprintf("%v %v", editor, tmpFile.Name()))
+	cmd.Stdin = os.Stdin
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+
+	err = cmd.Start()
+	checkErr(err, "start editor")
+
+	err = cmd.Wait()
+	checkErr(err, "wait for editor")
+
+	// read new code out of the file
+	contents, err := ioutil.ReadFile(tmpFile.Name())
+	checkErr(err, "read temp file")
+
+	function.Code = string(contents)
+
+	// upload the updated function
+	newfn, err := client.FunctionUpdate(function)
+	checkErr(err, "upload edited function")
+
+	fmt.Printf("function %v updated, new uuid: %v\n", newfn.Name, newfn.Uid)
+	return nil
+}

fission/httptrigger.go

@@ -26,7 +26,7 @@ import (
 	"github.com/satori/go.uuid"
 	"github.com/urfave/cli"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 // returns one of http.Method*
@@ -99,6 +99,22 @@ func htGet(c *cli.Context) error {
 }
 
 func htUpdate(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+	htName := c.String("name")
+	if len(htName) == 0 {
+		fatal("Need name of trigger, use --name")
+	}
+
+	ht, err := client.HTTPTriggerGet(&fission.Metadata{Name: htName})
+	checkErr(err, "get HTTP trigger")
+
+	newUid := c.String("uid")
+	ht.Function.Uid = newUid
+
+	_, err = client.HTTPTriggerUpdate(ht)
+	checkErr(err, "update HTTP trigger")
+
+	fmt.Printf("trigger '%v' updated\n", htName)
 	return nil
 }
 

fission/main.go

@@ -31,14 +31,19 @@ func main() {
 		cli.StringFlag{Name: "server", Usage: "Fission server URL", EnvVar: "FISSION_URL"},
 	}
 
+	// trigger method and url flags (used in function and route CLIs)
+	htMethodFlag := cli.StringFlag{Name: "method", Usage: "HTTP Method: GET|POST|PUT|DELETE|HEAD; defaults to GET"}
+	htUrlFlag := cli.StringFlag{Name: "url", Usage: "URL pattern (See gorilla/mux supported patterns)"}
+
 	// functions
 	fnNameFlag := cli.StringFlag{Name: "name", Usage: "function name"}
 	fnEnvNameFlag := cli.StringFlag{Name: "env", Usage: "environment name for function"}
 	fnCodeFlag := cli.StringFlag{Name: "code", Usage: "file containing source code, or - for stdin"}
 	fnUidFlag := cli.StringFlag{Name: "uid", Usage: "function uid, optional (use latest if unspecified)"}
 	fnSubcommands := []cli.Command{
-		{Name: "create", Usage: "Create new function", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag}, Action: fnCreate},
+		{Name: "create", Usage: "Create new function (and optionally, an HTTP route to it)", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, htUrlFlag, htMethodFlag}, Action: fnCreate},
 		{Name: "get", Usage: "Get function source code", Flags: []cli.Flag{fnNameFlag, fnUidFlag}, Action: fnGet},
+		{Name: "edit", Usage: "Edit function source code in $EDITOR", Flags: []cli.Flag{fnNameFlag, fnUidFlag}, Action: fnEdit},
 		{Name: "getmeta", Usage: "Get function metadata", Flags: []cli.Flag{fnNameFlag, fnUidFlag}, Action: fnGetMeta},
 		{Name: "update", Usage: "Update function source code", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag}, Action: fnUpdate},
 		{Name: "delete", Usage: "Delete function", Flags: []cli.Flag{fnNameFlag, fnUidFlag}, Action: fnDelete},
@@ -47,14 +52,12 @@ func main() {
 
 	// httptriggers
 	htNameFlag := cli.StringFlag{Name: "name", Usage: "HTTP Trigger name"}
-	htMethodFlag := cli.StringFlag{Name: "method", Usage: "HTTP Method: GET|POST|PUT|DELETE|HEAD; defaults to GET"}
-	htUrlFlag := cli.StringFlag{Name: "url", Usage: "URL pattern (see TODO for supported patterns)"}
 	htFnNameFlag := cli.StringFlag{Name: "function", Usage: "Function name"}
 	htFnUidFlag := cli.StringFlag{Name: "uid", Usage: "Function UID (optional; uses latest if unspecified)"}
 	htSubcommands := []cli.Command{
 		{Name: "create", Aliases: []string{"add"}, Usage: "Create HTTP trigger", Flags: []cli.Flag{htMethodFlag, htUrlFlag, htFnNameFlag, htFnUidFlag}, Action: htCreate},
 		{Name: "get", Usage: "Get HTTP trigger", Flags: []cli.Flag{htMethodFlag, htUrlFlag}, Action: htGet},
-		{Name: "update", Usage: "Update HTTP trigger", Flags: []cli.Flag{htMethodFlag, htUrlFlag, htFnNameFlag, htFnUidFlag}, Action: htUpdate},
+		{Name: "update", Usage: "Update HTTP trigger", Flags: []cli.Flag{htNameFlag, htFnNameFlag, htFnUidFlag}, Action: htUpdate},
 		{Name: "delete", Usage: "Delete HTTP trigger", Flags: []cli.Flag{htNameFlag}, Action: htDelete},
 		{Name: "list", Usage: "List HTTP triggers", Flags: []cli.Flag{}, Action: htList},
 	}
@@ -70,10 +73,26 @@ func main() {
 		{Name: "list", Usage: "List all environments", Flags: []cli.Flag{}, Action: envList},
 	}
 
+	// watches
+	wNameFlag := cli.StringFlag{Name: "name", Usage: "Watch name"}
+	wFnNameFlag := cli.StringFlag{Name: "function", Usage: "Function name"}
+	wFnUidFlag := cli.StringFlag{Name: "uid", Usage: "Function UID (optional; uses latest if unspecified)"}
+	wNamespaceFlag := cli.StringFlag{Name: "ns", Usage: "Namespace of resource to watch"}
+	wObjTypeFlag := cli.StringFlag{Name: "type", Usage: "Type of resource to watch (Pod, Service, etc.)"}
+	wLabelsFlag := cli.StringFlag{Name: "labels", Usage: "Label selector of the form a=b,c=d"}
+	wSubCommands := []cli.Command{
+		{Name: "create", Aliases: []string{"add"}, Usage: "Create a watch", Flags: []cli.Flag{wFnNameFlag, wFnUidFlag, wNamespaceFlag, wObjTypeFlag, wLabelsFlag}, Action: wCreate},
+		{Name: "get", Usage: "Get details about a watch", Flags: []cli.Flag{wNameFlag}, Action: wGet},
+		// TODO add update flag when supported
+		{Name: "delete", Usage: "Delete watch", Flags: []cli.Flag{wNameFlag}, Action: wDelete},
+		{Name: "list", Usage: "List all watches", Flags: []cli.Flag{}, Action: wList},
+	}
+
 	app.Commands = []cli.Command{
 		{Name: "function", Aliases: []string{"fn"}, Usage: "Create, update and manage functions", Subcommands: fnSubcommands},
 		{Name: "httptrigger", Aliases: []string{"ht", "route"}, Usage: "Manage HTTP triggers (routes) for functions", Subcommands: htSubcommands},
-		{Name: "environemnt", Aliases: []string{"env"}, Usage: "Manage environments", Subcommands: envSubcommands},
+		{Name: "environment", Aliases: []string{"env"}, Usage: "Manage environments", Subcommands: envSubcommands},
+		{Name: "watch", Aliases: []string{"w"}, Usage: "Manage watches", Subcommands: wSubCommands},
 
 		// Misc commands
 		{

fission/watch.go

@@ -0,0 +1,120 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"fmt"
+	"os"
+	"text/tabwriter"
+
+	"github.com/satori/go.uuid"
+	"github.com/urfave/cli"
+
+	"github.com/fission/fission"
+)
+
+func wCreate(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+
+	fnName := c.String("function")
+	if len(fnName) == 0 {
+		fatal("Need a function name to create a watch, use --function")
+	}
+	fnUid := c.String("uid")
+
+	namespace := c.String("ns")
+	if len(namespace) == 0 {
+		fmt.Println("Watch 'default' namespace. Use --ns <namespace> to override.")
+		namespace = "default"
+	}
+
+	objType := c.String("type")
+	if len(objType) == 0 {
+		fmt.Println("Object type unspecified, will watch pods.  Use --type <type> to override.")
+		objType = "pod"
+	}
+
+	labels := c.String("labels")
+	// empty 'labels' selects everything
+	if len(labels) == 0 {
+		fmt.Printf("Watching all objects of type '%v', use --labels to refine selection.\n", objType)
+	}
+
+	// automatically name watches
+	watchName := uuid.NewV4().String()
+
+	w := &fission.Watch{
+		Metadata: fission.Metadata{
+			Name: watchName,
+		},
+		Function: fission.Metadata{
+			Name: fnName,
+			Uid:  fnUid,
+		},
+		Namespace:     namespace,
+		ObjType:       objType,
+		LabelSelector: labels,
+		FieldSelector: "", // TODO
+	}
+
+	_, err := client.WatchCreate(w)
+	checkErr(err, "create watch")
+
+	fmt.Printf("watch '%v' created\n", w.Metadata.Name)
+	return err
+}
+
+func wGet(c *cli.Context) error {
+	return nil
+}
+
+func wUpdate(c *cli.Context) error {
+	return nil
+}
+
+func wDelete(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+
+	wName := c.String("name")
+	if len(wName) == 0 {
+		fatal("Need name of watch to delete, use --name")
+	}
+
+	err := client.WatchDelete(&fission.Metadata{Name: wName})
+	checkErr(err, "delete watch")
+
+	fmt.Printf("watch '%v' deleted\n", wName)
+	return nil
+}
+
+func wList(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+
+	ws, err := client.WatchList()
+	checkErr(err, "list watches")
+
+	w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0)
+
+	fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n", "NAME", "NAMESPACE", "OBJTYPE", "LABELS", "FUNCTION_NAME", "FUNCTION_UID")
+	for _, wa := range ws {
+		fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n",
+			wa.Metadata.Name, wa.Namespace, wa.ObjType, wa.LabelSelector, wa.Function.Name, wa.Function.Uid)
+	}
+	w.Flush()
+
+	return nil
+}

glide.lock

@@ -0,0 +1,270 @@
+hash: fb2f38693c5f2fade3db9434c266743d0f9ea9b5f878b44e79763e248136dcf2
+updated: 2016-11-14T01:06:12.764919869-08:00
+imports:
+- name: github.com/blang/semver
+  version: 60ec3488bfea7cca02b021d106d9911120d25fe9
+- name: github.com/coreos/etcd
+  version: 952eb4fadeef3f840bd7557544338c985d4ed0ef
+  subpackages:
+  - client
+  - pkg/pathutil
+  - pkg/types
+- name: github.com/coreos/go-oidc
+  version: 16c5ecc505f1efa0fe4685826fd9962c4d137e87
+  subpackages:
+  - http
+  - jose
+  - key
+  - oauth2
+  - oidc
+- name: github.com/coreos/pkg
+  version: fa29b1d70f0beaddd4c7021607cc3c3be8ce94b8
+  subpackages:
+  - capnslog
+  - health
+  - httputil
+  - timeutil
+- name: github.com/davecgh/go-spew
+  version: 346938d642f2ec3594ed81d874461961cd0faa76
+  subpackages:
+  - spew
+- name: github.com/dchest/uniuri
+  version: 8902c56451e9b58ff940bbe5fec35d5f9c04584a
+- name: github.com/docker/distribution
+  version: 6e62b398420f8c0444e34e8a69a350c139f65956
+  subpackages:
+  - digest
+  - reference
+- name: github.com/docopt/docopt-go
+  version: 784ddc588536785e7299f7272f39101f7faccc3f
+- name: github.com/emicklei/go-restful
+  version: 3d66f886316ac990eb502aaa89ea38546420b8b7
+  subpackages:
+  - log
+  - swagger
+- name: github.com/ghodss/yaml
+  version: 73d445a93680fa1a78ae23a5839bad48f32ba1ee
+- name: github.com/go-openapi/jsonpointer
+  version: 8d96a2dc61536b690bd36b2e9df0b3c0b62825b2
+- name: github.com/go-openapi/jsonreference
+  version: 36d33bfe519efae5632669801b180bf1a245da3b
+- name: github.com/go-openapi/spec
+  version: d1c18b339aece4b16ead6d253b85b6ad7180ea54
+- name: github.com/go-openapi/swag
+  version: 3b6d86cd965820f968760d5d419cb4add096bdd7
+- name: github.com/gogo/protobuf
+  version: 8d70fb3182befc465c4a1eac8ad4d38ff49778e2
+  subpackages:
+  - proto
+  - sortkeys
+- name: github.com/golang/glog
+  version: 23def4e6c14b4da8ac2ed8007337bc5eb5007998
+- name: github.com/golang/protobuf
+  version: 4bd1920723d7b7c925de087aa32e2187708897f7
+  subpackages:
+  - jsonpb
+  - proto
+- name: github.com/google/gofuzz
+  version: fd52762d25a41827db7ef64c43756fd4b9f7e382
+- name: github.com/gorilla/context
+  version: 08b5f424b9271eedf6f9f0ce86cb9396ed337a42
+- name: github.com/gorilla/handlers
+  version: ee54c7b44cab12289237fb8631314790076e728b
+- name: github.com/gorilla/mux
+  version: 0eeaf8392f5b04950925b8a69fe70f110fa7cbfc
+- name: github.com/jonboulle/clockwork
+  version: 2eee05ed794112d45db504eb05aa693efd2b8b09
+- name: github.com/juju/ratelimit
+  version: 77ed1c8a01217656d2080ad51981f6e99adaa177
+- name: github.com/mailru/easyjson
+  version: 159cdb893c982e3d1bc6450322fedd514f9c9de3
+  subpackages:
+  - buffer
+  - jlexer
+  - jwriter
+- name: github.com/pborman/uuid
+  version: 3d4f2ba23642d3cfd06bd4b54cf03d99d95c0f1b
+- name: github.com/PuerkitoBio/purell
+  version: 8a290539e2e8629dbc4e6bad948158f790ec31f4
+- name: github.com/PuerkitoBio/urlesc
+  version: 5bd2802263f21d8788851d5305584c82a5c75d7e
+- name: github.com/satori/go.uuid
+  version: 879c5887cd475cd7864858769793b2ceb0d44feb
+- name: github.com/Sirupsen/logrus
+  version: d26492970760ca5d33129d2d799e34be5c4782eb
+- name: github.com/spf13/pflag
+  version: 08b1a584251b5b62f458943640fc8ebd4d50aaa5
+- name: github.com/ugorji/go
+  version: f1f1a805ed361a0e078bb537e4ea78cd37dcf065
+  subpackages:
+  - codec
+- name: github.com/urfave/cli
+  version: a14d7d367bc02b1f57d88de97926727f2d936387
+- name: golang.org/x/net
+  version: 6acef71eb69611914f7a30939ea9f6e194c78172
+  subpackages:
+  - context
+  - http2
+  - http2/hpack
+  - idna
+- name: golang.org/x/oauth2
+  version: 045497edb6234273d67dbc25da3f2ddbc4c4cacf
+  subpackages:
+  - google
+  - internal
+  - jws
+  - jwt
+- name: golang.org/x/sys
+  version: b699b7032584f0953262cb2788a0ca19bb494703
+  subpackages:
+  - unix
+- name: golang.org/x/text
+  version: a263ba8db058568bb9beba166777d9c9dbe75d68
+  subpackages:
+  - cases
+  - internal
+  - internal/tag
+  - language
+  - runes
+  - secure/bidirule
+  - secure/precis
+  - transform
+  - unicode/bidi
+  - unicode/norm
+  - width
+- name: google.golang.org/appengine
+  version: 12d5545dc1cfa6047a286d5e853841b6471f4c19
+  subpackages:
+  - internal
+  - internal/app_identity
+  - internal/base
+  - internal/datastore
+  - internal/log
+  - internal/modules
+  - internal/remote_api
+  - internal/urlfetch
+  - urlfetch
+- name: google.golang.org/cloud
+  version: 975617b05ea8a58727e6c1a06b6161ff4185a9f2
+  subpackages:
+  - compute/metadata
+  - internal
+  - internal/opts
+  - storage
+- name: gopkg.in/inf.v0
+  version: 3887ee99ecf07df5b447e9b00d9c0b2adaa9f3e4
+- name: gopkg.in/yaml.v2
+  version: 53feefa2559fb8dfa8d81baad31be332c97d6c77
+- name: k8s.io/client-go
+  version: 843f7c4f28b1f647f664f883697107d5c02c5acc
+  subpackages:
+  - 1.4/kubernetes
+  - 1.4/pkg/api
+  - 1.4/pkg/api/v1
+  - 1.4/pkg/apis/extensions/v1beta1
+  - 1.4/pkg/labels
+  - 1.4/pkg/util/intstr
+  - 1.4/rest
+  - 1.5/discovery
+  - 1.5/kubernetes
+  - 1.5/kubernetes/typed/apps/v1alpha1
+  - 1.5/kubernetes/typed/authentication/v1beta1
+  - 1.5/kubernetes/typed/authorization/v1beta1
+  - 1.5/kubernetes/typed/autoscaling/v1
+  - 1.5/kubernetes/typed/batch/v1
+  - 1.5/kubernetes/typed/certificates/v1alpha1
+  - 1.5/kubernetes/typed/core/v1
+  - 1.5/kubernetes/typed/extensions/v1beta1
+  - 1.5/kubernetes/typed/policy/v1alpha1
+  - 1.5/kubernetes/typed/rbac/v1alpha1
+  - 1.5/kubernetes/typed/storage/v1beta1
+  - 1.5/pkg/api
+  - 1.5/pkg/api/errors
+  - 1.5/pkg/api/install
+  - 1.5/pkg/api/meta
+  - 1.5/pkg/api/meta/metatypes
+  - 1.5/pkg/api/resource
+  - 1.5/pkg/api/unversioned
+  - 1.5/pkg/api/v1
+  - 1.5/pkg/api/validation/path
+  - 1.5/pkg/apimachinery
+  - 1.5/pkg/apimachinery/announced
+  - 1.5/pkg/apimachinery/registered
+  - 1.5/pkg/apis/apps
+  - 1.5/pkg/apis/apps/install
+  - 1.5/pkg/apis/apps/v1alpha1
+  - 1.5/pkg/apis/authentication
+  - 1.5/pkg/apis/authentication/install
+  - 1.5/pkg/apis/authentication/v1beta1
+  - 1.5/pkg/apis/authorization
+  - 1.5/pkg/apis/authorization/install
+  - 1.5/pkg/apis/authorization/v1beta1
+  - 1.5/pkg/apis/autoscaling
+  - 1.5/pkg/apis/autoscaling/install
+  - 1.5/pkg/apis/autoscaling/v1
+  - 1.5/pkg/apis/batch
+  - 1.5/pkg/apis/batch/install
+  - 1.5/pkg/apis/batch/v1
+  - 1.5/pkg/apis/batch/v2alpha1
+  - 1.5/pkg/apis/certificates
+  - 1.5/pkg/apis/certificates/install
+  - 1.5/pkg/apis/certificates/v1alpha1
+  - 1.5/pkg/apis/extensions
+  - 1.5/pkg/apis/extensions/install
+  - 1.5/pkg/apis/extensions/v1beta1
+  - 1.5/pkg/apis/policy
+  - 1.5/pkg/apis/policy/install
+  - 1.5/pkg/apis/policy/v1alpha1
+  - 1.5/pkg/apis/rbac
+  - 1.5/pkg/apis/rbac/install
+  - 1.5/pkg/apis/rbac/v1alpha1
+  - 1.5/pkg/apis/storage
+  - 1.5/pkg/apis/storage/install
+  - 1.5/pkg/apis/storage/v1beta1
+  - 1.5/pkg/auth/user
+  - 1.5/pkg/conversion
+  - 1.5/pkg/conversion/queryparams
+  - 1.5/pkg/fields
+  - 1.5/pkg/genericapiserver/openapi/common
+  - 1.5/pkg/labels
+  - 1.5/pkg/runtime
+  - 1.5/pkg/runtime/serializer
+  - 1.5/pkg/runtime/serializer/json
+  - 1.5/pkg/runtime/serializer/protobuf
+  - 1.5/pkg/runtime/serializer/recognizer
+  - 1.5/pkg/runtime/serializer/streaming
+  - 1.5/pkg/runtime/serializer/versioning
+  - 1.5/pkg/selection
+  - 1.5/pkg/third_party/forked/golang/reflect
+  - 1.5/pkg/types
+  - 1.5/pkg/util
+  - 1.5/pkg/util/cert
+  - 1.5/pkg/util/clock
+  - 1.5/pkg/util/errors
+  - 1.5/pkg/util/flowcontrol
+  - 1.5/pkg/util/framer
+  - 1.5/pkg/util/integer
+  - 1.5/pkg/util/intstr
+  - 1.5/pkg/util/json
+  - 1.5/pkg/util/labels
+  - 1.5/pkg/util/net
+  - 1.5/pkg/util/parsers
+  - 1.5/pkg/util/rand
+  - 1.5/pkg/util/runtime
+  - 1.5/pkg/util/sets
+  - 1.5/pkg/util/uuid
+  - 1.5/pkg/util/validation
+  - 1.5/pkg/util/validation/field
+  - 1.5/pkg/util/wait
+  - 1.5/pkg/util/yaml
+  - 1.5/pkg/version
+  - 1.5/pkg/watch
+  - 1.5/pkg/watch/versioned
+  - 1.5/plugin/pkg/client/auth
+  - 1.5/plugin/pkg/client/auth/gcp
+  - 1.5/plugin/pkg/client/auth/oidc
+  - 1.5/rest
+  - 1.5/tools/clientcmd/api
+  - 1.5/tools/metrics
+  - 1.5/transport
+testImports: []

glide.yaml

@@ -0,0 +1,31 @@
+package: github.com/fission/fission
+import:
+- package: github.com/Sirupsen/logrus
+  version: ^0.11.0
+- package: github.com/coreos/etcd
+  subpackages:
+  - client
+- package: github.com/dchest/uniuri
+- package: github.com/docopt/docopt-go
+  version: ^0.6.2
+- package: github.com/gorilla/handlers
+  version: ^1.1.0
+- package: github.com/gorilla/mux
+  version: ^1.1.0
+- package: github.com/satori/go.uuid
+  version: ^1.1.0
+- package: github.com/urfave/cli
+  version: ^1.18.1
+- package: golang.org/x/net
+  subpackages:
+  - context
+- package: k8s.io/client-go
+  version: v1.5.0
+  subpackages:
+  - 1.5/kubernetes
+  - 1.5/pkg/api
+  - 1.5/pkg/api/v1
+  - 1.5/pkg/apis/extensions/v1beta1
+  - 1.5/pkg/labels
+  - 1.5/pkg/util/intstr
+  - 1.5/rest

hack/verify-gofmt.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+
+set -o errexit
+set -o nounset
+set -o pipefail
+
+find_files() {
+  find . -not \( \
+      \( \
+        -wholename '*/vendor/*' \
+      \) -prune \
+    \) -name '*.go'
+}
+
+GOFMT="gofmt -s"
+bad_files=$(find_files | grep -v '.glide/cache' | xargs $GOFMT -l)
+if [[ -n "${bad_files}" ]]; then
+  echo "!!! '$GOFMT' needs to be run on the following files: "
+  echo "${bad_files}"
+  exit 1
+fi

kubewatcher/kubewatcher.go

@@ -0,0 +1,203 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kubewatcher
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"strings"
+	"sync/atomic"
+
+	"k8s.io/client-go/1.5/kubernetes"
+	"k8s.io/client-go/1.5/pkg/api"
+	"k8s.io/client-go/1.5/pkg/runtime"
+	"k8s.io/client-go/1.5/pkg/watch"
+
+	"github.com/fission/fission"
+)
+
+type requestType int
+
+const (
+	SYNC requestType = iota
+)
+
+type (
+	KubeWatcher struct {
+		watches          map[string]watchSubscription
+		kubernetesClient *kubernetes.Clientset
+		requestChannel   chan *kubeWatcherRequest
+		publisher        Publisher
+	}
+
+	watchSubscription struct {
+		fission.Watch
+		kubeWatch watch.Interface
+		stopped   *int32
+	}
+
+	kubeWatcherRequest struct {
+		requestType
+		watches         []fission.Watch
+		responseChannel chan *kubeWatcherResponse
+	}
+	kubeWatcherResponse struct {
+		error
+	}
+)
+
+func MakeKubeWatcher(kubernetesClient *kubernetes.Clientset, publisher Publisher) *KubeWatcher {
+	kw := &KubeWatcher{
+		watches:          make(map[string]watchSubscription),
+		kubernetesClient: kubernetesClient,
+		publisher:        publisher,
+		requestChannel:   make(chan *kubeWatcherRequest),
+	}
+	go kw.svc()
+	return kw
+}
+
+func (kw *KubeWatcher) Sync(watches []fission.Watch) error {
+	req := &kubeWatcherRequest{
+		requestType:     SYNC,
+		watches:         watches,
+		responseChannel: make(chan *kubeWatcherResponse),
+	}
+	kw.requestChannel <- req
+	resp := <-req.responseChannel
+	return resp.error
+}
+
+func (kw *KubeWatcher) svc() {
+	for {
+		req := <-kw.requestChannel
+		switch req.requestType {
+		case SYNC:
+			newWatchUids := make(map[string]bool)
+			for _, w := range req.watches {
+				newWatchUids[w.Metadata.Uid] = true
+			}
+			// Remove old watches
+			for uid, ws := range kw.watches {
+				if _, ok := newWatchUids[uid]; !ok {
+					kw.removeWatch(&ws.Watch)
+				}
+			}
+			// Add new watches
+			for _, w := range req.watches {
+				if _, ok := kw.watches[w.Metadata.Uid]; !ok {
+					kw.addWatch(&w)
+				}
+			}
+			req.responseChannel <- &kubeWatcherResponse{error: nil}
+		}
+	}
+}
+
+// TODO lifted from kubernetes/pkg/kubectl/resource_printer.go.
+func printKubernetesObject(obj runtime.Object, w io.Writer) error {
+	switch obj := obj.(type) {
+	case *runtime.Unknown:
+		var buf bytes.Buffer
+		err := json.Indent(&buf, obj.Raw, "", "    ")
+		if err != nil {
+			return err
+		}
+		buf.WriteRune('\n')
+		_, err = buf.WriteTo(w)
+		return err
+	}
+
+	data, err := json.MarshalIndent(obj, "", "    ")
+	if err != nil {
+		return err
+	}
+	data = append(data, '\n')
+	_, err = w.Write(data)
+	return err
+}
+
+func (kw *KubeWatcher) createKubernetesWatch(w *fission.Watch) (watch.Interface, error) {
+	var wi watch.Interface
+	var err error
+
+	listOptions := api.ListOptions{} // TODO populate labelselector and fieldselector
+
+	// TODO handle the full list of types
+	switch strings.ToUpper(w.ObjType) {
+	case "POD":
+		wi, err = kw.kubernetesClient.Core().Pods(w.Namespace).Watch(listOptions)
+	case "SERVICE":
+		wi, err = kw.kubernetesClient.Core().Services(w.Namespace).Watch(listOptions)
+	default:
+		msg := fmt.Sprintf("Error: unknown obj type '%v'", w.ObjType)
+		log.Println(msg)
+		err = errors.New(msg)
+	}
+	return wi, err
+}
+
+func (kw *KubeWatcher) addWatch(w *fission.Watch) error {
+	log.Printf("Adding watch %v: %v", w.Metadata.Name, w.Function.Name)
+	wi, err := kw.createKubernetesWatch(w)
+	if err != nil {
+		return err
+	}
+	var stopped int32 = 0
+	ws := &watchSubscription{
+		Watch:     *w,
+		kubeWatch: wi,
+		stopped:   &stopped,
+	}
+	kw.watches[w.Metadata.Uid] = *ws
+	go ws.eventDispatchLoop(kw.publisher)
+	return nil
+}
+
+func (kw *KubeWatcher) removeWatch(w *fission.Watch) error {
+	log.Printf("Removing watch %v: %v", w.Metadata.Name, w.Function.Name)
+	ws, ok := kw.watches[w.Metadata.Uid]
+	if !ok {
+		return fission.MakeError(fission.ErrorNotFound,
+			fmt.Sprintf("watch doesn't exist: %v", w.Metadata))
+	}
+	delete(kw.watches, w.Metadata.Uid)
+	atomic.StoreInt32(ws.stopped, 1)
+	ws.kubeWatch.Stop()
+	return nil
+}
+
+func (ws *watchSubscription) eventDispatchLoop(publisher Publisher) {
+	log.Println("Listening to watch ", ws.Watch.Metadata.Name)
+	for {
+		ev, more := <-ws.kubeWatch.ResultChan()
+		if !more {
+			log.Println("Watch stopped", ws.Watch.Metadata.Name)
+			break
+		}
+		publisher.Publish(ev, ws.Watch.Target)
+	}
+	if atomic.LoadInt32(ws.stopped) != 0 {
+		// TODO can this happen?  How do we start the watch again from the right
+		// point?
+		log.Panicf("Watch channel closed unexpectedly")
+	}
+}

kubewatcher/main.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kubewatcher
+
+import (
+	"log"
+
+	"k8s.io/client-go/1.5/kubernetes"
+	"k8s.io/client-go/1.5/rest"
+
+	"github.com/fission/fission/controller/client"
+)
+
+// Get a kubernetes client using the pod's service account.
+func getKubernetesClient() (*kubernetes.Clientset, error) {
+	// creates the in-cluster config
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		log.Printf("Error getting kubernetes client config: %v", err)
+		return nil, err
+	}
+
+	// creates the clientset
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		log.Printf("Error getting kubernetes client: %v", err)
+		return nil, err
+	}
+
+	return clientset, nil
+}
+
+func Start(controllerUrl string, routerUrl string) error {
+	kubeClient, err := getKubernetesClient()
+	if err != nil {
+		return err
+	}
+	poster := MakeWebhookPublisher(routerUrl)
+	kubeWatch := MakeKubeWatcher(kubeClient, poster)
+
+	client := client.MakeClient(controllerUrl)
+	MakeWatchSync(client, kubeWatch)
+
+	return nil
+}

kubewatcher/publisher.go

@@ -0,0 +1,30 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kubewatcher
+
+import (
+	"k8s.io/client-go/1.5/pkg/watch"
+)
+
+type (
+	Publisher interface {
+		// Publish an event to a "target".  Target's meaning depends on the
+		// publisher: it's a URL in the case of a webhook publisher, or a queue
+		// name in a queue-based publisher such as NATS.
+		Publish(event watch.Event, target string)
+	}
+)

kubewatcher/watchSync.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kubewatcher
+
+import (
+	"log"
+	"time"
+
+	"github.com/fission/fission/controller/client"
+)
+
+type (
+	WatchSync struct {
+		client      *client.Client
+		kubeWatcher *KubeWatcher
+	}
+)
+
+func MakeWatchSync(client *client.Client, kubeWatcher *KubeWatcher) *WatchSync {
+	ws := &WatchSync{
+		client:      client,
+		kubeWatcher: kubeWatcher,
+	}
+	go ws.syncSvc()
+	return ws
+}
+
+func (ws *WatchSync) syncSvc() {
+	failureCount := 0
+	maxFailures := 6
+	for {
+		watches, err := ws.client.WatchList()
+		if err != nil {
+			failureCount++
+			if failureCount > maxFailures {
+				log.Fatalf("Failed to connect to controller: %v", err)
+			}
+			time.Sleep(10 * time.Second)
+			continue
+		}
+
+		ws.kubeWatcher.Sync(watches)
+		time.Sleep(3 * time.Second)
+	}
+}

kubewatcher/webhookPublisher.go

@@ -0,0 +1,135 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kubewatcher
+
+import (
+	"bytes"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"reflect"
+	"strings"
+	"time"
+
+	"k8s.io/client-go/1.5/pkg/watch"
+)
+
+type (
+	// A webhook publisher for a single URL. Satisifies the Publisher interface.
+	WebhookPublisher struct {
+		requestChannel chan *publishRequest
+
+		maxRetries int
+		retryDelay time.Duration
+
+		baseUrl string
+	}
+	publishRequest struct {
+		url        string
+		watchEvent watch.Event
+		retries    int
+		retryDelay time.Duration
+	}
+)
+
+func MakeWebhookPublisher(baseUrl string) *WebhookPublisher {
+	p := &WebhookPublisher{
+		baseUrl:        baseUrl,
+		requestChannel: make(chan *publishRequest, 32), // buffered channel
+		// TODO make this configurable
+		maxRetries: 10,
+		retryDelay: 500 * time.Millisecond,
+	}
+	go p.svc()
+	return p
+}
+
+func (p *WebhookPublisher) Publish(watchEvent watch.Event, url string) {
+	p.requestChannel <- &publishRequest{
+		watchEvent: watchEvent,
+		url:        url,
+		retries:    p.maxRetries,
+		retryDelay: p.retryDelay,
+	}
+}
+
+func (p *WebhookPublisher) svc() {
+	for {
+		r := <-p.requestChannel
+		p.makeHttpRequest(r)
+	}
+}
+
+func (p *WebhookPublisher) makeHttpRequest(r *publishRequest) {
+
+	url := p.baseUrl + "/" + strings.TrimPrefix(r.url, "/")
+	log.Printf("Making HTTP request to %v", url)
+
+	// Serialize the object
+	var buf bytes.Buffer
+	err := printKubernetesObject(r.watchEvent.Object, &buf)
+	if err != nil {
+		log.Printf("Failed to serialize object: %v", err)
+		// TODO send a POST request indicating error
+	}
+
+	// Create request
+	req, err := http.NewRequest("POST", url, &buf)
+	if err != nil {
+		log.Printf("Failed to create request to %v", url)
+		// can't do anything more, drop the event.
+		return
+	}
+
+	// Event and object type aren't in the serialized object
+	req.Header.Add("Content-Type", "application/json")
+	req.Header.Add("X-Kubernetes-Event-Type", string(r.watchEvent.Type))
+	req.Header.Add("X-Kubernetes-Object-Type", reflect.TypeOf(r.watchEvent.Object).Elem().Name())
+
+	// Make the request
+	resp, err := http.DefaultClient.Do(req)
+
+	// All done if the request succeeded with 200 OK.
+	if err == nil && resp.StatusCode == 200 {
+		resp.Body.Close()
+		return
+	}
+
+	// Log errors
+	if err != nil {
+		log.Printf("Request failed: %v", r)
+	} else if resp.StatusCode != 200 {
+		log.Printf("Request returned failure: %v", resp.StatusCode)
+		body, err := ioutil.ReadAll(resp.Body)
+		resp.Body.Close()
+		if err == nil {
+			log.Printf("request error: %v", string(body))
+		}
+	}
+
+	// Schedule a retry, or give up if out of retries
+	r.retries--
+	if r.retries > 0 {
+		r.retryDelay *= time.Duration(2)
+		time.AfterFunc(r.retryDelay, func() {
+			p.requestChannel <- r
+		})
+	} else {
+		log.Printf("Final retry failed, giving up on %v", url)
+		// Event dropped
+	}
+}

poolmgr/api.go

@@ -29,9 +29,9 @@ import (
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
-	"github.com/platform9/fission/cache"
+	"github.com/fission/fission/cache"
-	controllerclient "github.com/platform9/fission/controller/client"
+	controllerclient "github.com/fission/fission/controller/client"
 )
 
 type funcSvc struct {
@@ -119,11 +119,19 @@ func (api *API) getFunctionEnv(m *fission.Metadata) (*fission.Environment, error
 }
 
 func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) {
-	// Check function -> svc map
+	// Make sure we have the full metadata.  This ensures that
+	// poolmgr does not implicitly interpret empty-UID as latest
+	// version.
+	if len(m.Uid) == 0 {
+		return "", fission.MakeError(fission.ErrorInvalidArgument,
+			fmt.Sprintf("invalid metadata for function %v", m.Name))
+	}
+
+	// Check function -> svc cache
 	log.Printf("[%v] Checking for cached function service", m.Name)
 	fsvc, err := api.fsCache.GetByFunction(m)
 	if err == nil {
-		// Cached, return svc name
+		// Cached, return svc address
 		return fsvc.address, nil
 	}
 

poolmgr/cleanup.go

@@ -0,0 +1,147 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package poolmgr
+
+import (
+	"log"
+	"time"
+
+	"k8s.io/client-go/1.5/kubernetes"
+	"k8s.io/client-go/1.5/pkg/api"
+)
+
+// cleanupOldPoolmgrResources looks for resources created by an old
+// poolmgr instance and cleans them up.
+func cleanupOldPoolmgrResources(client *kubernetes.Clientset, namespace string, instanceId string) {
+	go func() {
+		err := cleanup(client, namespace, instanceId)
+		if err != nil {
+			// TODO retry cleanup; logged and ignored for now
+			log.Printf("Failed to cleanup: %v", err)
+		}
+	}()
+}
+
+func cleanup(client *kubernetes.Clientset, namespace string, instanceId string) error {
+	// Deployments are used for idle pools and can be cleaned up
+	// immediately.  (We should "adopt" these instead of creating
+	// a new pool.)
+	err := cleanupDeployments(client, namespace, instanceId)
+	if err != nil {
+		return err
+	}
+	// See K8s #33845 and related bugs: deleting a deployment
+	// through the API doesn't cause the associated ReplicaSet to
+	// be deleted.  (Fixed recently, but we may be running a
+	// version before the fix.)
+	err = cleanupReplicaSets(client, namespace, instanceId)
+	if err != nil {
+		return err
+	}
+
+	// Pods might still be running user functions, so we give them
+	// a few minutes before terminating them.  This time is the
+	// maximum function runtime, plus the time a router might
+	// still route to an old instance, i.e. router cache expiry
+	// time.
+	time.Sleep(6 * time.Minute)
+
+	err = cleanupPods(client, namespace, instanceId)
+	if err != nil {
+		return err
+	}
+
+	err = cleanupServices(client, namespace, instanceId)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func cleanupDeployments(client *kubernetes.Clientset, namespace string, instanceId string) error {
+	deploymentList, err := client.Extensions().Deployments(namespace).List(api.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, dep := range deploymentList.Items {
+		id, ok := dep.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL]
+		if ok && id != instanceId {
+			log.Printf("Cleaning up deployment %v", dep.ObjectMeta.Name)
+			err := client.Extensions().Deployments(namespace).Delete(dep.ObjectMeta.Name, nil)
+			logErr("cleaning up deployment", err)
+			// ignore err
+		}
+	}
+	return nil
+}
+
+func cleanupReplicaSets(client *kubernetes.Clientset, namespace string, instanceId string) error {
+	rsList, err := client.Extensions().ReplicaSets(namespace).List(api.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, rs := range rsList.Items {
+		id, ok := rs.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL]
+		if ok && id != instanceId {
+			log.Printf("Cleaning up replicaset %v", rs.ObjectMeta.Name)
+			err := client.Extensions().ReplicaSets(namespace).Delete(rs.ObjectMeta.Name, nil)
+			logErr("cleaning up replicaset", err)
+		}
+	}
+	return nil
+}
+
+func cleanupPods(client *kubernetes.Clientset, namespace string, instanceId string) error {
+	podList, err := client.Core().Pods(namespace).List(api.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, pod := range podList.Items {
+		id, ok := pod.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL]
+		if ok && id != instanceId {
+			log.Printf("Cleaning up pod %v", pod.ObjectMeta.Name)
+			err := client.Core().Pods(namespace).Delete(pod.ObjectMeta.Name, nil)
+			logErr("cleaning up pod", err)
+			// ignore err
+		}
+	}
+	return nil
+}
+
+func cleanupServices(client *kubernetes.Clientset, namespace string, instanceId string) error {
+	svcList, err := client.Core().Services(namespace).List(api.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, svc := range svcList.Items {
+		id, ok := svc.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL]
+		if ok && id != instanceId {
+			log.Printf("Cleaning up svc %v", svc.ObjectMeta.Name)
+			err := client.Core().Services(namespace).Delete(svc.ObjectMeta.Name, nil)
+			logErr("cleaning up svc", err)
+			// ignore err
+		}
+	}
+	return nil
+}
+
+func logErr(msg string, err error) {
+	if err != nil {
+		log.Printf("Error %v: %v", msg, err)
+	}
+}

poolmgr/client/client.go

@@ -22,7 +22,7 @@ import (
 
 	"bytes"
 	"encoding/json"
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 	"io/ioutil"
 	"net/url"
 )

poolmgr/functionServiceCache.go

@@ -20,8 +20,8 @@ import (
 	"log"
 	"time"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
-	"github.com/platform9/fission/cache"
+	"github.com/fission/fission/cache"
 )
 
 type fscRequestType int

poolmgr/functionServiceCache_test.go

@@ -4,7 +4,7 @@ import (
 	"log"
 	"testing"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 	"time"
 )
 
@@ -58,15 +58,19 @@ func TestFunctionServiceCache(t *testing.T) {
 		log.Panicf("Failed to touch fsvc: %v", err)
 	}
 
-	err = fsc.DeleteByPod(fsvc.podName)
+	deleted, err := fsc.DeleteByPod(fsvc.podName, 0)
 	if err != nil {
 		fsc.Log()
 		log.Panicf("Failed to delete fsvc: %v", err)
 	}
+	if !deleted {
+		fsc.Log()
+		log.Panicf("Did not delete fsvc")
+	}
 
 	_, err = fsc.GetByFunction(fsvc.function)
 	if err == nil {
 		fsc.Log()
-		log.Panicf("found fsvc while expecting empty cache", err)
+		log.Panicf("found fsvc while expecting empty cache: %v", err)
 	}
 }

poolmgr/gp.go

@@ -29,29 +29,33 @@ import (
 	"time"
 
 	"github.com/dchest/uniuri"
-	"k8s.io/client-go/1.4/kubernetes"
+	"k8s.io/client-go/1.5/kubernetes"
-	"k8s.io/client-go/1.4/pkg/api"
+	"k8s.io/client-go/1.5/pkg/api"
-	"k8s.io/client-go/1.4/pkg/api/v1"
+	"k8s.io/client-go/1.5/pkg/api/v1"
-	"k8s.io/client-go/1.4/pkg/apis/extensions/v1beta1"
+	"k8s.io/client-go/1.5/pkg/apis/extensions/v1beta1"
-	"k8s.io/client-go/1.4/pkg/labels"
+	"k8s.io/client-go/1.5/pkg/labels"
-	"k8s.io/client-go/1.4/pkg/util/intstr"
+	"k8s.io/client-go/1.5/pkg/util/intstr"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
+const POOLMGR_INSTANCEID_LABEL string = "poolmgrInstanceId"
+
 type (
 	GenericPool struct {
 		env              *fission.Environment
-		replicas         int32               // num containers
+		replicas         int32               // num idle pods
 		deployment       *v1beta1.Deployment // kubernetes deployment
 		namespace        string              // namespace to keep our resources
 		podReadyTimeout  time.Duration       // timeout for generic pods to become ready
 		controllerUrl    string
 		idlePodReapTime  time.Duration         // pods unused for idlePodReapTime are deleted
 		fsCache          *functionServiceCache // cache funcSvc's by function, address and podname
-		useSvc           bool                  // create service
+		useSvc           bool                  // create k8s service for specialized pods
 		poolInstanceId   string                // small random string to uniquify pod names
 		kubernetesClient *kubernetes.Clientset
+		instanceId       string // poolmgr instance id
+		labelsForPool    map[string]string
 		requestChannel   chan *choosePodRequest
 	}
 
@@ -72,7 +76,8 @@ func MakeGenericPool(
 	env *fission.Environment,
 	initialReplicas int32,
 	namespace string,
-	fsCache *functionServiceCache) (*GenericPool, error) {
+	fsCache *functionServiceCache,
+	instanceId string) (*GenericPool, error) {
 
 	log.Printf("Creating pool for environment %v", env.Metadata)
 	// TODO: in general we need to provide the user a way to configure pools.  Initial
@@ -88,8 +93,16 @@ func MakeGenericPool(
 		idlePodReapTime:  3 * time.Minute, // TODO make this configurable
 		fsCache:          fsCache,
 		poolInstanceId:   uniuri.NewLen(8),
+		instanceId:       instanceId,
 
-		useSvc: false,
+		useSvc: false, // defaults off -- svc takes a second or more to become routable, slowing cold start
+	}
+
+	// Labels for generic deployment/RS/pods.
+	gp.labelsForPool = map[string]string{
+		"environmentName":        gp.env.Metadata.Name,
+		"environmentUid":         gp.env.Metadata.Uid,
+		POOLMGR_INSTANCEID_LABEL: gp.instanceId,
 	}
 
 	// create the pool
@@ -200,33 +213,37 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*v1.Pod, error)
 	}
 }
 
-func labelsForMetadata(metadata *fission.Metadata) map[string]string {
+func (gp *GenericPool) labelsForFunction(metadata *fission.Metadata) map[string]string {
 	return map[string]string{
-		"functionName": metadata.Name,
+		"functionName":           metadata.Name,
-		"functionUid":  metadata.Uid,
+		"functionUid":            metadata.Uid,
-		"unmanaged":    "true", // this allows us to easily find pods not managed by the deployment
+		"unmanaged":              "true", // this allows us to easily find pods not managed by the deployment
+		POOLMGR_INSTANCEID_LABEL: gp.instanceId,
 	}
 }
 
+func (gp *GenericPool) scheduleDeletePod(name string) {
+	go func() {
+		// The sleep allows debugging or collecting logs from the pod before it's
+		// cleaned up.  (We need a better solutions for both those things; log
+		// aggregation and storage will help.)
+		log.Printf("Error in pod '%v', scheduling cleanup", name)
+		time.Sleep(5 * time.Minute)
+		gp.kubernetesClient.Core().Pods(gp.namespace).Delete(name, nil)
+	}()
+}
+
 // specializePod chooses a pod, copies the required user-defined function to that pod
 // (via fetcher), and calls the function-run container to load it, resulting in a
 // specialized pod.
-func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error) {
+func (gp *GenericPool) specializePod(pod *v1.Pod, metadata *fission.Metadata) error {
-	newLabels := labelsForMetadata(metadata)
-
-	log.Printf("[%v] Choosing pod from pool", metadata)
-	pod, err := gp.choosePod(newLabels)
-	if err != nil {
-		return nil, err
-	}
-
 	// for fetcher we don't need to create a service, just talk to the pod directly
 	podIP := pod.Status.PodIP
 	if len(podIP) == 0 {
-		return nil, errors.New("Pod has no IP")
+		return errors.New("Pod has no IP")
 	}
 
-	// tell fetcher to get the function
+	// tell fetcher to get the function.
 	fetcherUrl := fmt.Sprintf("http://%v:8000/", podIP)
 	functionUrl := fmt.Sprintf("%v/v1/functions/%v?uid=%v&raw=1",
 		gp.controllerUrl, metadata.Name, metadata.Uid)
@@ -236,11 +253,11 @@ func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error
 	resp, err := http.Post(fetcherUrl, "application/json", bytes.NewReader([]byte(fetcherRequest)))
 	if err != nil {
 		// TODO we should retry this call in case fetcher hasn't come up yet
-		return nil, err
+		return err
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode != 200 {
-		return nil, errors.New(fmt.Sprintf("Error from fetcher: %v", resp.Status))
+		return errors.New(fmt.Sprintf("Error from fetcher: %v", resp.Status))
 	}
 
 	// get function run container to specialize
@@ -252,8 +269,9 @@ func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error
 	for i := 0; i < maxRetries; i++ {
 		resp2, err := http.Post(specializeUrl, "text/plain", bytes.NewReader([]byte{}))
 		if err == nil && resp2.StatusCode < 300 {
+			// Success
 			resp2.Body.Close()
-			return pod, nil
+			return nil
 		}
 
 		// Only retry for the specific case of a connection error.
@@ -273,10 +291,10 @@ func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error
 			err = fission.MakeErrorFromHTTP(resp2)
 		}
 		log.Printf("Failed to specialize pod: %v", err)
-		return nil, err
+		return err
 	}
 
-	return pod, nil
+	return nil
 }
 
 // A pool is a deployment of generic containers for an env.  This
@@ -285,31 +303,24 @@ func (gp *GenericPool) createPool() error {
 	poolDeploymentName := fmt.Sprintf("%v-%v-%v",
 		gp.env.Metadata.Name, gp.env.Metadata.Uid, strings.ToLower(gp.poolInstanceId))
 
-	podLabels := map[string]string{
-		"pool": poolDeploymentName,
-	}
-
 	sharedMountPath := "/userfunc"
 	deployment := &v1beta1.Deployment{
 		ObjectMeta: v1.ObjectMeta{
-			Name: poolDeploymentName,
+			Name:   poolDeploymentName,
-			Labels: map[string]string{
+			Labels: gp.labelsForPool,
-				"environmentName": gp.env.Metadata.Name,
-				"environmentUid":  gp.env.Metadata.Uid,
-			},
 		},
 		Spec: v1beta1.DeploymentSpec{
 			Replicas: &gp.replicas,
 			Selector: &v1beta1.LabelSelector{
-				MatchLabels: podLabels,
+				MatchLabels: gp.labelsForPool,
 			},
 			Template: v1.PodTemplateSpec{
 				ObjectMeta: v1.ObjectMeta{
-					Labels: podLabels,
+					Labels: gp.labelsForPool,
 				},
 				Spec: v1.PodSpec{
 					Volumes: []v1.Volume{
-						v1.Volume{
+						{
 							Name: "userfunc",
 							VolumeSource: v1.VolumeSource{
 								EmptyDir: &v1.EmptyDirVolumeSource{},
@@ -317,25 +328,25 @@ func (gp *GenericPool) createPool() error {
 						},
 					},
 					Containers: []v1.Container{
-						v1.Container{
+						{
 							Name:                   gp.env.Metadata.Name,
 							Image:                  gp.env.RunContainerImageUrl,
 							ImagePullPolicy:        v1.PullIfNotPresent,
 							TerminationMessagePath: "/dev/termination-log",
 							VolumeMounts: []v1.VolumeMount{
-								v1.VolumeMount{
+								{
 									Name:      "userfunc",
 									MountPath: sharedMountPath,
 								},
 							},
 						},
-						v1.Container{
+						{
 							Name:                   "fetcher",
 							Image:                  "fission/fetcher",
 							ImagePullPolicy:        v1.PullIfNotPresent,
 							TerminationMessagePath: "/dev/termination-log",
 							VolumeMounts: []v1.VolumeMount{
-								v1.VolumeMount{
+								{
 									Name:      "userfunc",
 									MountPath: sharedMountPath,
 								},
@@ -384,7 +395,7 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*v1.Ser
 		Spec: v1.ServiceSpec{
 			Type: v1.ServiceTypeClusterIP,
 			Ports: []v1.ServicePort{
-				v1.ServicePort{
+				{
 					Protocol:   v1.ProtocolTCP,
 					Port:       80,
 					TargetPort: intstr.FromInt(8888),
@@ -398,10 +409,19 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*v1.Ser
 }
 
 func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) {
-	pod, err := gp.specializePod(m)
+
+	log.Printf("[%v] Choosing pod from pool", m)
+	newLabels := gp.labelsForFunction(m)
+	pod, err := gp.choosePod(newLabels)
 	if err != nil {
 		return nil, err
 	}
+
+	err = gp.specializePod(pod, m)
+	if err != nil {
+		gp.scheduleDeletePod(pod.ObjectMeta.Name)
+		return nil, err
+	}
 	log.Printf("Specialized pod: %v", pod.ObjectMeta.Name)
 
 	var svcHost string
@@ -411,12 +431,14 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) {
 			svcName += ("-" + m.Uid)
 		}
 
-		labels := labelsForMetadata(m)
+		labels := gp.labelsForFunction(m)
 		svc, err := gp.createSvc(svcName, labels)
 		if err != nil {
+			gp.scheduleDeletePod(pod.ObjectMeta.Name)
 			return nil, err
 		}
 		if svc.ObjectMeta.Name != svcName {
+			gp.scheduleDeletePod(pod.ObjectMeta.Name)
 			return nil, errors.New(fmt.Sprintf("sanity check failed for svc %v", svc.ObjectMeta.Name))
 		}
 
@@ -443,7 +465,9 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) {
 		// our own.  TODO: this is grossly inefficient, improve it with some sort of state
 		// machine
 		log.Printf("func svc already exists: %v", existingFsvc.podName)
-		go gp.CleanupFunctionService(fsvc.podName)
+		go func() {
+			gp.kubernetesClient.Core().Pods(gp.namespace).Delete(fsvc.podName, nil)
+		}()
 		return existingFsvc, nil
 	}
 	return fsvc, nil
@@ -487,3 +511,43 @@ func (gp *GenericPool) idlePodReaper() {
 		}
 	}
 }
+
+// destroys the pool -- the deployment, replicaset and pods
+func (gp *GenericPool) destroy() error {
+	// Destroy deployment
+	err := gp.kubernetesClient.Extensions().Deployments(gp.namespace).Delete(gp.deployment.ObjectMeta.Name, nil)
+	if err != nil {
+		log.Printf("Error destroying deployment: %v", err)
+		return err
+	}
+
+	// Destroy ReplicaSet.  Pre-1.6 K8s versions don't do this
+	// automatically but post-1.6 K8s will, and may beat us to it,
+	// so don't error out if we fail.
+	rsList, err := gp.kubernetesClient.Extensions().ReplicaSets(gp.namespace).List(api.ListOptions{
+		LabelSelector: labels.Set(gp.labelsForPool).AsSelector(),
+	})
+	if len(rsList.Items) >= 0 {
+		for _, rs := range rsList.Items {
+			err = gp.kubernetesClient.Extensions().ReplicaSets(gp.namespace).Delete(rs.ObjectMeta.Name, nil)
+			if err != nil {
+				log.Printf("Error deleting replicaset, ignoring: %v", err)
+			}
+		}
+	}
+
+	// Destroy Pods.  See note above.
+	podList, err := gp.kubernetesClient.Core().Pods(gp.namespace).List(api.ListOptions{
+		LabelSelector: labels.Set(gp.labelsForPool).AsSelector(),
+	})
+	if len(podList.Items) >= 0 {
+		for _, pod := range podList.Items {
+			err = gp.kubernetesClient.Core().Pods(gp.namespace).Delete(pod.ObjectMeta.Name, nil)
+			if err != nil {
+				log.Printf("Error deleting pod, ignoring: %v", err)
+			}
+		}
+	}
+
+	return nil
+}

poolmgr/gpm.go

@@ -20,10 +20,17 @@ import (
 	"log"
 	"time"
 
-	"k8s.io/client-go/1.4/kubernetes"
+	"k8s.io/client-go/1.5/kubernetes"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
-	"github.com/platform9/fission/controller/client"
+	"github.com/fission/fission/controller/client"
+)
+
+type requestType int
+
+const (
+	GET_POOL requestType = iota
+	CLEANUP_POOLS
 )
 
 type (
@@ -34,11 +41,13 @@ type (
 		controllerUrl    string
 		controllerClient *client.Client
 		fsCache          *functionServiceCache
-
+		instanceId       string
-		requestChannel chan *request
+		requestChannel   chan *request
 	}
 	request struct {
+		requestType
 		env             *fission.Environment
+		envList         []fission.Environment
 		responseChannel chan *response
 	}
 	response struct {
@@ -47,7 +56,13 @@ type (
 	}
 )
 
-func MakeGenericPoolManager(controllerUrl string, kubernetesClient *kubernetes.Clientset, namespace string, fsCache *functionServiceCache) *GenericPoolManager {
+func MakeGenericPoolManager(
+	controllerUrl string,
+	kubernetesClient *kubernetes.Clientset,
+	namespace string,
+	fsCache *functionServiceCache,
+	instanceId string) *GenericPoolManager {
+
 	gpm := &GenericPoolManager{
 		pools:            make(map[fission.Environment]*GenericPool),
 		kubernetesClient: kubernetesClient,
@@ -55,6 +70,7 @@ func MakeGenericPoolManager(controllerUrl string, kubernetesClient *kubernetes.C
 		controllerUrl:    controllerUrl,
 		controllerClient: client.MakeClient(controllerUrl),
 		fsCache:          fsCache,
+		instanceId:       instanceId,
 		requestChannel:   make(chan *request),
 	}
 	go gpm.service()
@@ -65,12 +81,16 @@ func MakeGenericPoolManager(controllerUrl string, kubernetesClient *kubernetes.C
 
 func (gpm *GenericPoolManager) service() {
 	for {
-		select {
+		req := <-gpm.requestChannel
-		case req := <-gpm.requestChannel:
+		switch req.requestType {
+		case GET_POOL:
 			var err error
 			pool, ok := gpm.pools[*req.env]
 			if !ok {
-				pool, err = MakeGenericPool(gpm.controllerUrl, gpm.kubernetesClient, req.env, 3, gpm.namespace, gpm.fsCache)
+				pool, err = MakeGenericPool(
+					gpm.controllerUrl, gpm.kubernetesClient, req.env,
+					3, // TODO configurable/autoscalable
+					gpm.namespace, gpm.fsCache, gpm.instanceId)
 				if err != nil {
 					req.responseChannel <- &response{error: err}
 					continue
@@ -78,17 +98,45 @@ func (gpm *GenericPoolManager) service() {
 				gpm.pools[*req.env] = pool
 			}
 			req.responseChannel <- &response{pool: pool}
+		case CLEANUP_POOLS:
+			uids := make(map[string]bool)
+			for _, env := range req.envList {
+				uids[env.Metadata.Uid] = true
+			}
+			for env, pool := range gpm.pools {
+				_, ok := uids[env.Metadata.Uid]
+				if !ok {
+					// Env no longer exists -- remove our cache
+					log.Printf("Destroying generic pool for environment [%v]", env)
+					delete(gpm.pools, env)
+
+					// and delete the pool asynchronously.
+					go pool.destroy()
+				}
+			}
+			// no response, caller doesn't wait
 		}
 	}
 }
 
 func (gpm *GenericPoolManager) GetPool(env *fission.Environment) (*GenericPool, error) {
 	c := make(chan *response)
-	gpm.requestChannel <- &request{env: env, responseChannel: c}
+	gpm.requestChannel <- &request{
+		requestType:     GET_POOL,
+		env:             env,
+		responseChannel: c,
+	}
 	resp := <-c
 	return resp.pool, resp.error
 }
 
+func (gpm *GenericPoolManager) CleanupPools(envs []fission.Environment) {
+	gpm.requestChannel <- &request{
+		requestType: CLEANUP_POOLS,
+		envList:     envs,
+	}
+}
+
 func (gpm *GenericPoolManager) eagerPoolCreator() {
 	failureCount := 0
 	maxFailures := 5
@@ -115,5 +163,8 @@ func (gpm *GenericPoolManager) eagerPoolCreator() {
 				log.Printf("eager-create pool failed: %v", err)
 			}
 		}
+
+		// Clean up pools whose env was deleted
+		gpm.CleanupPools(envs)
 	}
 }

poolmgr/poolmgr.go

@@ -20,10 +20,11 @@ import (
 	"log"
 	"strings"
 
-	"k8s.io/client-go/1.4/kubernetes"
+	"github.com/dchest/uniuri"
-	"k8s.io/client-go/1.4/rest"
+	"k8s.io/client-go/1.5/kubernetes"
+	"k8s.io/client-go/1.5/rest"
 
-	controllerclient "github.com/platform9/fission/controller/client"
+	controllerclient "github.com/fission/fission/controller/client"
 )
 
 // Get a kubernetes client using the pod's service account.  This only
@@ -56,11 +57,13 @@ func StartPoolmgr(controllerUrl string, namespace string, port int) error {
 		return err
 	}
 
+	instanceId := uniuri.NewLen(8)
+	cleanupOldPoolmgrResources(kubernetesClient, namespace, instanceId)
+
 	fsCache := MakeFunctionServiceCache()
+	gpm := MakeGenericPoolManager(controllerUrl, kubernetesClient, namespace, fsCache, instanceId)
 
-	gpm := MakeGenericPoolManager(controllerUrl, kubernetesClient, namespace, fsCache)
 	api := MakeAPI(gpm, controllerClient, fsCache)
-
 	go api.Serve(port)
 
 	return nil

resource.go

@@ -1,3 +1,19 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package fission
 
 func (f Function) Key() string {
@@ -11,3 +27,7 @@ func (e Environment) Key() string {
 func (ht HTTPTrigger) Key() string {
 	return ht.Metadata.Name
 }
+
+func (w Watch) Key() string {
+	return w.Metadata.Name
+}

router/functionHandler.go

@@ -25,8 +25,8 @@ import (
 	"net/url"
 	"time"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
-	poolmgrClient "github.com/platform9/fission/poolmgr/client"
+	poolmgrClient "github.com/fission/fission/poolmgr/client"
 )
 
 type functionHandler struct {

router/functionHandler_test.go

@@ -23,7 +23,7 @@ import (
 	"net/url"
 	"testing"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 func createBackendService(testResponseString string) *url.URL {

router/functionServiceMap.go

@@ -21,8 +21,8 @@ import (
 	"net/url"
 	"time"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
-	"github.com/platform9/fission/cache"
+	"github.com/fission/fission/cache"
 )
 
 type functionServiceMap struct {

router/functionServiceMap_test.go

@@ -20,7 +20,7 @@ import (
 	"net/url"
 	"testing"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 func TestFunctionServiceMap(t *testing.T) {

router/httpTriggers.go

@@ -18,15 +18,13 @@ package router
 
 import (
 	"log"
-	"os"
-	"strconv"
 	"time"
 
 	"github.com/gorilla/mux"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
-	controllerClient "github.com/platform9/fission/controller/client"
+	controllerClient "github.com/fission/fission/controller/client"
-	poolmgrClient "github.com/platform9/fission/poolmgr/client"
+	poolmgrClient "github.com/fission/fission/poolmgr/client"
 )
 
 type HTTPTriggerSet struct {
@@ -35,6 +33,7 @@ type HTTPTriggerSet struct {
 	controller *controllerClient.Client
 	poolmgr    *poolmgrClient.Client
 	triggers   []fission.HTTPTrigger
+	functions  []fission.Function
 }
 
 func makeHTTPTriggerSet(fmap *functionServiceMap, controller *controllerClient.Client, poolmgr *poolmgrClient.Client) *HTTPTriggerSet {
@@ -49,20 +48,45 @@ func makeHTTPTriggerSet(fmap *functionServiceMap, controller *controllerClient.C
 
 func (ts *HTTPTriggerSet) subscribeRouter(mr *mutableRouter) {
 	ts.mutableRouter = mr
-	mr.updateRouter(ts.getRouterFromTriggers())
+	mr.updateRouter(ts.getRouter())
 	go ts.watchTriggers()
 }
 
-func (ts *HTTPTriggerSet) getRouterFromTriggers() *mux.Router {
+func (ts *HTTPTriggerSet) getRouter() *mux.Router {
 	muxRouter := mux.NewRouter()
+
+	// make a name -> latest version map
+	latestVersions := make(map[string]string)
+	for _, f := range ts.functions {
+		latestVersions[f.Metadata.Name] = f.Metadata.Uid
+	}
+
+	// HTTP triggers setup by the user
 	for _, trigger := range ts.triggers {
+		m := trigger.Function
+		if len(m.Uid) == 0 {
+			// explicitly use the latest function version
+			m.Uid = latestVersions[m.Name]
+		}
 		fh := &functionHandler{
 			fmap:     ts.functionServiceMap,
-			Function: trigger.Function,
+			Function: m,
 			poolmgr:  ts.poolmgr,
 		}
-		muxRouter.HandleFunc(trigger.UrlPattern, fh.handler)
+		muxRouter.HandleFunc(trigger.UrlPattern, fh.handler).Methods(trigger.Method)
 	}
+
+	// Internal triggers for (the latest version of) each function
+	for _, function := range ts.functions {
+		m := fission.Metadata{Name: function.Metadata.Name}
+		fh := &functionHandler{
+			fmap:     ts.functionServiceMap,
+			Function: function.Metadata,
+			poolmgr:  ts.poolmgr,
+		}
+		muxRouter.HandleFunc(fission.UrlForFunction(&m), fh.handler)
+	}
+
 	return muxRouter
 }
 
@@ -72,25 +96,10 @@ func (ts *HTTPTriggerSet) watchTriggers() {
 	}
 
 	// the number of connection failures we'll accept before quitting
-	var err error
 	maxFailures := 5
-	maxFailuresEnv := os.Getenv("FISSION_ROUTER_MAX_FAILURES")
-	if len(maxFailuresEnv) != 0 {
-		maxFailures, err = strconv.Atoi(maxFailuresEnv)
-		if err != nil {
-			log.Fatalf("FISSION_ROUTER_MAX_FAILURES must be an integer, found %v", maxFailuresEnv)
-		}
-	}
 
 	// amount of time to sleep between polling calls
-	pollSleepSec := 3
+	pollSleepDuration := 3 * time.Second
-	pollSleepEnv := os.Getenv("FISSION_ROUTER_POLL_SLEEP_SECONDS")
-	if len(pollSleepEnv) != 0 {
-		pollSleepSec, err = strconv.Atoi(pollSleepEnv)
-		if err != nil {
-			log.Fatalf("FISSION_ROUTER_POLL_SLEEP_SECONDS must be an integer, found %v", pollSleepEnv)
-		}
-	}
 
 	// Watch controller for updates to triggers and update the router accordingly.
 	// TODO change this to use a watch API; or maybe even watch etcd directly.
@@ -102,9 +111,18 @@ func (ts *HTTPTriggerSet) watchTriggers() {
 			if failureCount >= maxFailures {
 				log.Fatalf("Failed to connect to controller after %v retries: %v", failureCount, err)
 			}
+			time.Sleep(pollSleepDuration)
+			continue
 		}
 		ts.triggers = triggers
-		ts.mutableRouter.updateRouter(ts.getRouterFromTriggers())
+
-		time.Sleep(time.Duration(pollSleepSec) * time.Second)
+		functions, err := ts.controller.FunctionList()
+		if err != nil {
+			log.Fatalf("Failed to get function list")
+		}
+		ts.functions = functions
+
+		ts.mutableRouter.updateRouter(ts.getRouter())
+		time.Sleep(pollSleepDuration)
 	}
 }

router/router.go

@@ -43,12 +43,14 @@ import (
 	"fmt"
 	"log"
 	"net/http"
+	"os"
 	"time"
 
+	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
 
-	controllerClient "github.com/platform9/fission/controller/client"
+	controllerClient "github.com/fission/fission/controller/client"
-	poolmgrClient "github.com/platform9/fission/poolmgr/client"
+	poolmgrClient "github.com/fission/fission/poolmgr/client"
 )
 
 // request url ---[mux]---> Function(name,uid) ----[fmap]----> k8s service url
@@ -65,7 +67,7 @@ func router(httpTriggerSet *HTTPTriggerSet) *mutableRouter {
 func serve(port int, httpTriggerSet *HTTPTriggerSet) {
 	mr := router(httpTriggerSet)
 	url := fmt.Sprintf(":%v", port)
-	http.ListenAndServe(url, mr)
+	http.ListenAndServe(url, handlers.LoggingHandler(os.Stdout, mr))
 }
 
 func Start(port int, controllerUrl string, poolmgrUrl string) {

router/router_test.go

@@ -21,7 +21,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/platform9/fission"
+	"github.com/fission/fission"
 )
 
 func TestRouter(t *testing.T) {
@@ -35,7 +35,7 @@ func TestRouter(t *testing.T) {
 
 	triggers := makeHTTPTriggerSet(fmap, nil, nil)
 	triggerUrl := "/foo"
-	triggers.triggers = append(triggers.triggers, fission.HTTPTrigger{UrlPattern: triggerUrl, Function: *fn})
+	triggers.triggers = append(triggers.triggers, fission.HTTPTrigger{UrlPattern: triggerUrl, Function: *fn, Method: "GET"})
 
 	port := 4242
 	go serve(port, triggers)

types.go

@@ -53,6 +53,20 @@ type (
 		Function   Metadata `json:"function"`
 	}
 
+	// Watch is a specification of Kubernetes watch along with a URL to post events to.
+	Watch struct {
+		Metadata `json:"metadata"`
+
+		Namespace     string `json:"namespace"`
+		ObjType       string `json:"objtype"`
+		LabelSelector string `json:"labelselector"`
+		FieldSelector string `json:"fieldselector"`
+
+		Function Metadata `json:"function"`
+
+		Target string `json:"target"` // Watch publish target (URL, NATS stream, etc)
+	}
+
 	// Errors returned by the Fission API.
 	Error struct {
 		Code    errorCode `json:"code"`
@@ -69,4 +83,5 @@ const (
 	ErrorNameExists
 	ErrorInvalidArgument
 	ErrorNoSpace
+	ErrorNotImplmented
 )