func eventIngestBulkMeta(
	index outil.Selector,
	pipelineSel *outil.Selector,
	data outputs.Data,
) interface{} {
	type bulkMetaIndex struct {
		Index    string `json:"_index"`
		DocType  string `json:"_type"`
		Pipeline string `json:"pipeline"`
	}
	type bulkMeta struct {
		Index bulkMetaIndex `json:"index"`
	}

	event := data.Event
	pipeline, _ := pipelineSel.Select(event)
	if pipeline == "" {
		return eventBulkMeta(index, nil, data)
	}

	return bulkMeta{
		Index: bulkMetaIndex{
			Index:    getIndex(event, index),
			Pipeline: pipeline,
			DocType:  event["type"].(string),
		},
	}
}

func makePublishRPUSH(conn redis.Conn, key outil.Selector) (publishFn, error) {
	if !key.IsConst() {
		// TODO: more clever bulk handling batching events with same key
		return publishEventsPipeline(conn, "RPUSH"), nil
	}

	var major, minor int
	var versionRaw [][]byte

	respRaw, err := conn.Do("INFO")
	resp, err := redis.Bytes(respRaw, err)
	if err != nil {
		return nil, err
	}

	versionRaw = versionRegex.FindSubmatch(resp)
	if versionRaw == nil {
		err = errors.New("unable to read redis_version")
		return nil, err
	}

	major, err = strconv.Atoi(string(versionRaw[1]))
	if err != nil {
		return nil, err
	}

	minor, err = strconv.Atoi(string(versionRaw[2]))
	if err != nil {
		return nil, err
	}

	// Check Redis version number choosing the method
	// how RPUSH shall be used. With version 2.4 RPUSH
	// can accept multiple values at once turning RPUSH
	// into batch like call instead of relying on pipelining.
	//
	// Versions 1.0 to 2.3 only accept one value being send with
	// RPUSH requiring pipelining.
	//
	// See: http://redis.io/commands/rpush
	multiValue := major > 2 || (major == 2 && minor >= 4)
	if multiValue {
		return publishEventsBulk(conn, key, "RPUSH"), nil
	}
	return publishEventsPipeline(conn, "RPUSH"), nil
}

func publishEventsBulk(conn redis.Conn, key outil.Selector, command string) publishFn {
	// XXX: requires key.IsConst() == true
	dest, _ := key.Select(common.MapStr{})
	return func(_ outil.Selector, data []outputs.Data) ([]outputs.Data, error) {
		args := make([]interface{}, 1, len(data)+1)
		args[0] = dest

		data, args = serializeEvents(args, 1, data)
		if (len(args) - 1) == 0 {
			return nil, nil
		}

		// RPUSH returns total length of list -> fail and retry all on error
		_, err := conn.Do(command, args...)
		if err != nil {
			logp.Err("Failed to %v to redis list (%v) with %v", command, err)
			return data, err
		}

		return nil, nil
	}
}

// getIndex returns the full index name
// Index is either defined in the config as part of the output
// or can be overload by the event through setting index
func getIndex(event common.MapStr, index outil.Selector) string {

	ts := time.Time(event["@timestamp"].(common.Time)).UTC()

	// Check for dynamic index
	// XXX: is this used/needed?
	if _, ok := event["beat"]; ok {
		beatMeta, ok := event["beat"].(common.MapStr)
		if ok {
			// Check if index is set dynamically
			if dynamicIndex, ok := beatMeta["index"]; ok {
				if dynamicIndexValue, ok := dynamicIndex.(string); ok {
					return fmt.Sprintf("%s-%d.%02d.%02d",
						dynamicIndexValue, ts.Year(), ts.Month(), ts.Day())
				}
			}
		}
	}

	str, _ := index.Select(event)
	return str
}