func (s *Store) GetExecutions(jobName string) ([]*Execution, error) {
	prefix := fmt.Sprintf("%s/executions/%s", s.keyspace, jobName)
	res, err := s.Client.List(prefix)
	if err != nil {
		return nil, err
	}

	var executions []*Execution

	for _, node := range res {
		if store.Backend(s.backend) != store.ZK {
			path := store.SplitKey(node.Key)
			dir := path[len(path)-2]
			if dir != jobName {
				continue
			}
		}
		var execution Execution
		err := json.Unmarshal([]byte(node.Value), &execution)
		if err != nil {
			return nil, err
		}
		executions = append(executions, &execution)
	}
	return executions, nil
}

// AtomicPut put a value at "key" if the key has not been
// modified in the meantime, throws an error if this is the case
func (s *Zookeeper) AtomicPut(key string, value []byte, previous *store.KVPair, _ *store.WriteOptions) (bool, *store.KVPair, error) {
	var lastIndex uint64

	if previous != nil {
		meta, err := s.client.Set(s.normalize(key), value, int32(previous.LastIndex))
		if err != nil {
			// Compare Failed
			if err == zk.ErrBadVersion {
				return false, nil, store.ErrKeyModified
			}
			return false, nil, err
		}
		lastIndex = uint64(meta.Version)
	} else {
		// Interpret previous == nil as create operation.
		_, err := s.client.Create(s.normalize(key), value, 0, zk.WorldACL(zk.PermAll))
		if err != nil {
			// Directory does not exist
			if err == zk.ErrNoNode {

				// Create the directory
				parts := store.SplitKey(strings.TrimSuffix(key, "/"))
				parts = parts[:len(parts)-1]
				if err = s.createFullPath(parts, false); err != nil {
					// Failed to create the directory.
					return false, nil, err
				}

				// Create the node
				if _, err := s.client.Create(s.normalize(key), value, 0, zk.WorldACL(zk.PermAll)); err != nil {
					// Node exist error (when previous nil)
					if err == zk.ErrNodeExists {
						return false, nil, store.ErrKeyExists
					}
					return false, nil, err
				}

			} else {
				// Node Exists error (when previous nil)
				if err == zk.ErrNodeExists {
					return false, nil, store.ErrKeyExists
				}

				// Unhandled error
				return false, nil, err
			}
		}
		lastIndex = 0 // Newly created nodes have version 0.
	}

	pair := &store.KVPair{
		Key:       key,
		Value:     value,
		LastIndex: lastIndex,
	}

	return true, pair, nil
}

// Put a value at "key"
func (s *Zookeeper) Put(key string, value []byte, opts *store.WriteOptions) error {
	fkey := store.Normalize(key)

	exists, err := s.Exists(key)
	if err != nil {
		return err
	}

	if !exists {
		if opts != nil && opts.Ephemeral {
			s.createFullPath(store.SplitKey(key), opts.Ephemeral)
		} else {
			s.createFullPath(store.SplitKey(key), false)
		}
	}

	_, err = s.client.Set(fkey, value, -1)
	return err
}

// Put a value at "key"
func (s *Zookeeper) Put(key string, value []byte, opts *store.WriteOptions) error {
	fkey := s.normalize(key)

	exists, err := s.Exists(key)
	if err != nil {
		return err
	}

	if !exists {
		if opts != nil && opts.TTL > 0 {
			s.createFullPath(store.SplitKey(strings.TrimSuffix(key, "/")), true)
		} else {
			s.createFullPath(store.SplitKey(strings.TrimSuffix(key, "/")), false)
		}
	}

	_, err = s.client.Set(fkey, value, -1)
	return err
}