func (l *UdpInput) Run(output chan common.MapStr) error {
	logp.Info("[UdpInput] Running UDP Input")
	addr := net.UDPAddr{
		Port: l.Port,
		IP:   net.ParseIP("0.0.0.0"),
	}
	server, err := net.ListenUDP("udp", &addr)
	server.SetReadBuffer(1048576)

	if err != nil {
		logp.Err("couldn't start listening: " + err.Error())
		return nil
	}

	logp.Info("[UdpInput] Listening on port %d", l.Port)

	i := 0
	for {
		i++
		buf := make([]byte, 4096)
		rlen, addr, err := server.ReadFromUDP(buf)
		if err != nil {
			logp.Err("couldn't read from UDP: " + err.Error())
		}
		go l.handlePacket(buf, rlen, i, addr, output)
	}
	return nil
}

func (reader *ReaderType) Run(output chan common.MapStr) error {
	logp.Info("Attempting to start %d inputs", len(reader.Input))

	for _, plugin := range reader.Input {
		err := plugin.Run(output)
		if err != nil {
			logp.Err("Fail to start input plugin %s : %s", plugin.InputType(), err)
			return err
		} else {
			logp.Info("Started input plugin %s", plugin.InputType())
		}
	}
	return nil
}

func (l *TailInput) Init(config inputs.MothershipConfig) error {
	l.Type = "tail"
	l.Config = config
	l.FileName = config.Filename
	l.RollTime = 30 * time.Minute

	logp.Info("[TailInput] Initialized with file " + l.FileName)
	return nil
}

func (l *SyslogInput) Init(config inputs.MothershipConfig) error {

	l.Config = config
	if config.Port == 0 {
		return errors.New("No Input Port specified")
	}
	l.Port = config.Port

	if config.Type == "" {
		return errors.New("No Event Type specified")
	}
	l.Type = config.Type

	logp.Info("[SyslogInput] Using Port %d", l.Port)
	logp.Info("[SyslogInput] Adding Event Type %s", l.Type)

	return nil
}

func (l *ProcfsInput) Init(config inputs.MothershipConfig) error {

	l.Config = config

	l.Tick_interval = config.Tick_interval

	logp.Info("[ProcfsInput] Initialized, using tick interval " + strconv.Itoa(l.Tick_interval))

	return nil
}

func writeHeapProfile(filename string) {
	f, err := os.Create(filename)
	if err != nil {
		logp.Err("Failed creating file %s: %s", filename, err)
		return
	}
	pprof.WriteHeapProfile(f)
	f.Close()

	logp.Info("Created memory profile file %s.", filename)
}

func (l *PackagesInput) doStuff(output chan common.MapStr) {

	now := func() time.Time {
		t := time.Now()
		return t
	}

	// construct event and write it to channel
	event := common.MapStr{}

	//text := "null event"
	//event["message"] = &text

	event["message"] = "packages event"
	event["type"] = l.Type

	event.EnsureTimestampField(now)
	event.EnsureCountField()

	/////////////
	cmd := exec.Command("/bin/rpm", "-qa", "--queryformat", "%{NAME}:::%{VERSION}:::%{ARCH}##")
	var out bytes.Buffer
	cmd.Stdout = &out
	err := cmd.Run()
	if err != nil {
		logp.Info("Error occurred")
		return
	}

	items := strings.Split(out.String(), "##")
	rpmList := make([]RPMPackage, 0)

	for _, line := range items {
		item := strings.Split(line, ":::")
		if len(item) < 3 {
			continue
		}

		pkg := RPMPackage{
			Name:    item[0],
			Version: item[1],
			Arch:    item[2],
		}

		rpmList = append(rpmList, pkg)
	}

	event["packages"] = rpmList
	output <- event

}

func (out *KafkaOutput) Init(config outputs.MothershipConfig, topology_expire int) error {

	if config.Host == "" {
		return errors.New("No Kafka brokers specified")
	}
	out.BrokerList = strings.Split(config.Host, ",")

	if config.Topic == "" {
		return errors.New("No Kafka topic specified")
	}
	out.Topic = config.Topic

	out.Timeout = 5 * time.Second
	if config.Timeout != 0 {
		out.Timeout = time.Duration(config.Timeout) * time.Second
	}

	out.FlushInterval = 1000 * time.Millisecond
	out.ReconnectInterval = time.Duration(1) * time.Second
	if config.Reconnect_interval != 0 {
		out.ReconnectInterval = time.Duration(config.Reconnect_interval) * time.Second
	}

	//sarama.Logger = log.New(os.Stdout, "[KafkaOutput]", log.LstdFlags)
	logp.Info("[KafkaOutput] Using Kafka brokers %s", config.Host)
	logp.Info("[KafkaOutput] Kafka connection timeout %s", out.Timeout)
	logp.Info("[KafkaOutput] Kafka reconnect interval %s", out.ReconnectInterval)
	logp.Info("[KafkaOutput] Kafka flushing interval %s", out.FlushInterval)
	logp.Info("[KafkaOutput] Publishing to topic %s", out.Topic)

	out.sendingQueue = make(chan KafkaQueueMsg, 1000)

	out.Reconnect()
	go out.SendMessagesGoroutine()

	return nil
}

func (l *TcpInput) Run(output chan common.MapStr) error {
	logp.Info("[TcpInput] Running TCP Input")
	server, err := net.Listen("tcp", ":"+strconv.Itoa(l.Port))
	if err != nil {
		logp.Err("couldn't start listening: " + err.Error())
		return nil
	}
	logp.Info("[TcpInput] Listening on port %d", l.Port)

	// dispatch the master listen thread
	go func(server net.Listener) {
		for {
			// accept incoming connections
			conn, err := server.Accept()
			if err != nil {
				logp.Err("Error accepting: ", err.Error())
			} else {
				// dispatch individual connection handler threads
				go l.handleConn(conn, output)
			}
		}
	}(server)
	return nil
}

func (l *StdinInput) doStuff(output chan common.MapStr) {
	reader := bufio.NewReader(os.Stdin)
	buffer := new(bytes.Buffer)

	var source string = fmt.Sprintf("%s:%s", os.Getenv("REMOTE_HOST"), os.Getenv("REMOTE_PORT"))
	var ssl_client_dn string = os.Getenv("SSL_CLIENT_DN")
	var offset int64 = 0
	var line uint64 = 0
	var read_timeout = 10 * time.Second

	logp.Debug("stdinput", "Handling New Connection from %s", source)

	now := func() time.Time {
		t := time.Now()
		return t
	}

	for {
		text, bytesread, err := l.readline(reader, buffer, read_timeout)

		if err != nil {
			logp.Info("Unexpected state reading from %v; error: %s\n", os.Getenv("SSL_CLIENT_DN"), err)
			return
		}

		logp.Debug("stdinputlines", "New Line: %s", &text)

		line++

		event := common.MapStr{}
		event["ssl_client_dn"] = &ssl_client_dn
		event["source"] = &source
		event["offset"] = offset
		event["line"] = line
		event["message"] = text
		event["type"] = l.Type

		event.EnsureTimestampField(now)
		event.EnsureCountField()

		offset += int64(bytesread)

		logp.Debug("stdinput", "InputEvent: %v", event)
		output <- event // ship the new event downstream
		os.Stdout.Write([]byte("OK"))
	}
	logp.Debug("stdinput", "Closed Connection from %s", source)
}

func LoadGeoIPData(config Geoip) *libgeo.GeoIP {

	geoip_paths := []string{
		"/usr/share/GeoIP/GeoIP.dat",
		"/usr/local/var/GeoIP/GeoIP.dat",
	}
	if config.Paths != nil {
		geoip_paths = *config.Paths
	}
	if len(geoip_paths) == 0 {
		// disabled
		return nil
	}

	// look for the first existing path
	var geoip_path string
	for _, path := range geoip_paths {
		fi, err := os.Lstat(path)
		if err != nil {
			continue
		}

		if fi.Mode()&os.ModeSymlink == os.ModeSymlink {
			// follow symlink
			geoip_path, err = filepath.EvalSymlinks(path)
			if err != nil {
				logp.Warn("Could not load GeoIP data: %s", err.Error())
				return nil
			}
		} else {
			geoip_path = path
		}
		break
	}

	if len(geoip_path) == 0 {
		logp.Warn("Couldn't load GeoIP database")
		return nil
	}

	geoLite, err := libgeo.Load(geoip_path)
	if err != nil {
		logp.Warn("Could not load GeoIP data: %s", err.Error())
	}

	logp.Info("Loaded GeoIP data from: %s", geoip_path)
	return geoLite
}

func (l *TcpInput) handleConn(client net.Conn, output chan common.MapStr) {
	reader := bufio.NewReader(client)
	buffer := new(bytes.Buffer)

	var source string = client.RemoteAddr().String()
	var offset int64 = 0
	var line uint64 = 0
	var read_timeout = 10 * time.Second

	logp.Debug("tcpinput", "Handling New Connection from %s", source)

	now := func() time.Time {
		t := time.Now()
		return t
	}

	for {
		text, bytesread, err := l.readline(reader, buffer, read_timeout)

		if err != nil {
			logp.Info("Unexpected state reading from %v; error: %s\n", client.RemoteAddr().String, err)
			return
		}

		logp.Debug("tcpinputlines", "New Line: %s", &text)

		line++

		event := common.MapStr{}
		event["source"] = &source
		event["offset"] = offset
		event["line"] = line
		event["message"] = text
		event["type"] = l.Type

		event.EnsureTimestampField(now)
		event.EnsureCountField()

		offset += int64(bytesread)

		logp.Debug("tcpinput", "InputEvent: %v", event)
		output <- event // ship the new event downstream
		client.Write([]byte("OK"))
	}
	logp.Debug("tcpinput", "Closed Connection from %s", source)
}

func newInputInstance(name string) inputs.InputInterface {
	logp.Info("creating new instance of type %s", name)
	switch name {
	case "tcp":
		return new(tcp.TcpInput)
	case "udp":
		return new(udp.UdpInput)
	case "tail":
		return new(tail.TailInput)
	case "syslog":
		return new(syslog.SyslogInput)
	case "procfs":
		return new(procfs.ProcfsInput)
	case "packages":
		return new(packages.PackagesInput)
	case "null":
		return new(null.NullInput)
	}
	return nil
}

func (reader *ReaderType) Init(inputMap map[string]inputs.MothershipConfig) error {
	logp.Info("reader input config", inputMap)

	var globalConf inputs.MothershipConfig

	for inputId, config := range inputMap {
		// default instance 0
		inputName, instance := inputId, "0"
		if strings.Contains(inputId, "_") {
			// otherwise grok tcp_2 as inputName = tcp, instance = 2
			sv := strings.Split(inputId, "_")
			inputName, instance = sv[0], sv[1]
		}
		logp.Info(fmt.Sprintf("input type: %s instance: %s\n", inputName, instance))
		logp.Debug("reader", "instance config: %s", config)

		// handling for "global" config section
		if inputName == "global" {
			logp.Info("global input configuration read")
			globalConf = config
		}

		plugin := newInputInstance(inputName)
		if plugin != nil && config.Enabled {
			config.Normalize(globalConf)
			err := plugin.Init(config)
			if err != nil {
				logp.Err("Fail to initialize %s plugin as input: %s", inputName, err)
				return err
			} else {
				logp.Info("Initialized %s plugin as input", inputName)
			}
			reader.Input = append(reader.Input, plugin)
		}
	}

	if len(reader.Input) == 0 {
		logp.Info("No inputs are defined. Please define one under the input section.")
		return errors.New("No input are defined. Please define one under the input section.")
	} else {
		logp.Info("%d inputs defined", len(reader.Input))
	}

	return nil
}

func DropPrivileges(config RunOptions) error {
	var err error

	if config.Uid == nil {
		// not found, no dropping privileges but no err
		return nil
	}

	if config.Gid == nil {
		return errors.New("GID must be specified for dropping privileges")
	}

	logp.Info("Switching to user: %d.%d", config.Uid, config.Gid)

	if err = syscall.Setgid(*config.Gid); err != nil {
		return fmt.Errorf("setgid: %s", err.Error())
	}

	if err = syscall.Setuid(*config.Uid); err != nil {
		return fmt.Errorf("setuid: %s", err.Error())
	}

	return nil
}

func (publisher *PublisherType) Init(outputs map[string]outputs.MothershipConfig, shipper ShipperConfig) error {
	var err error
	publisher.IgnoreOutgoing = shipper.Ignore_outgoing

	publisher.disabled = *publishDisabled
	if publisher.disabled {
		logp.Info("Dry run mode. All output types except the file based one are disabled.")
	}

	publisher.GeoLite = common.LoadGeoIPData(shipper.Geoip)

	for outputId, plugin := range EnabledOutputPlugins {
		outputName := outputId.String()
		output, exists := outputs[outputName]
		if exists && output.Enabled && !publisher.disabled {
			err := plugin.Init(output, shipper.Topology_expire)
			if err != nil {
				logp.Err("Fail to initialize %s plugin as output: %s", outputName, err)
				return err
			}
			publisher.Output = append(publisher.Output, plugin)

			if output.Save_topology {
				if publisher.TopologyOutput != nil {
					logp.Err("Multiple outputs defined to store topology. Please add save_topology = true option only for one output.")
					return errors.New("Multiple outputs defined to store topology")
				}
				publisher.TopologyOutput = plugin
				logp.Info("Using %s to store the topology", outputName)
			}
		}
	}

	if !publisher.disabled {
		if len(publisher.Output) == 0 {
			logp.Info("No outputs are defined. Please define one under the shipper->output section.")
			return errors.New("No outputs are defined. Please define one under the shipper->output section.")
		}

		if publisher.TopologyOutput == nil {
			logp.Warn("No output is defined to store the topology. The server fields might not be filled.")
		}
	}

	publisher.name = shipper.Name
	if len(publisher.name) == 0 {
		// use the hostname
		publisher.name, err = os.Hostname()
		if err != nil {
			return err
		}

		logp.Info("No shipper name configured, using hostname '%s'", publisher.name)
	}

	publisher.tags = shipper.Tags

	if !publisher.disabled && publisher.TopologyOutput != nil {
		RefreshTopologyFreq := 10 * time.Second
		if shipper.Refresh_topology_freq != 0 {
			RefreshTopologyFreq = time.Duration(shipper.Refresh_topology_freq) * time.Second
		}
		publisher.RefreshTopologyTimer = time.Tick(RefreshTopologyFreq)
		logp.Info("Topology map refreshed every %s", RefreshTopologyFreq)

		// register shipper and its public IP addresses
		err = publisher.PublishTopology()
		if err != nil {
			logp.Err("Failed to publish topology: %s", err)
			return err
		}

		// update topology periodically
		go publisher.UpdateTopologyPeriodically()
	}

	publisher.Queue = make(chan common.MapStr, 10000)
	go publisher.publishFromQueue()

	return nil
}

// Initialize Elasticsearch as output
func (out *ElasticsearchOutput) Init(config outputs.MothershipConfig, topology_expire int) error {

	if len(config.Protocol) == 0 {
		config.Protocol = "http"
	}

	var urls []string

	if len(config.Hosts) > 0 {
		// use hosts setting
		for _, host := range config.Hosts {
			url := fmt.Sprintf("%s://%s%s", config.Protocol, host, config.Path)
			urls = append(urls, url)
		}
	} else {
		// use host and port settings
		url := fmt.Sprintf("%s://%s:%d%s", config.Protocol, config.Host, config.Port, config.Path)
		urls = append(urls, url)
	}

	es := NewElasticsearch(urls, config.Username, config.Password)
	out.Conn = es

	if config.Index != "" {
		out.Index = config.Index
	} else {
		out.Index = "packetbeat"
	}

	out.TopologyExpire = 15000
	if topology_expire != 0 {
		out.TopologyExpire = topology_expire /*sec*/ * 1000 // millisec
	}

	out.FlushInterval = 1000 * time.Millisecond
	if config.Flush_interval != nil {
		out.FlushInterval = time.Duration(*config.Flush_interval) * time.Millisecond
	}
	out.BulkMaxSize = 10000
	if config.Bulk_size != nil {
		out.BulkMaxSize = *config.Bulk_size
	}

	if config.Max_retries != nil {
		out.Conn.SetMaxRetries(*config.Max_retries)
	}

	logp.Info("[ElasticsearchOutput] Using Elasticsearch %s", urls)
	logp.Info("[ElasticsearchOutput] Using index pattern [%s-]YYYY.MM.DD", out.Index)
	logp.Info("[ElasticsearchOutput] Topology expires after %ds", out.TopologyExpire/1000)
	if out.FlushInterval > 0 {
		logp.Info("[ElasticsearchOutput] Insert events in batches. Flush interval is %s. Bulk size is %d.", out.FlushInterval, out.BulkMaxSize)
	} else {
		logp.Info("[ElasticsearchOutput] Insert events one by one. This might affect the performance of the shipper.")
	}

	if config.Save_topology {
		err := out.EnableTTL()
		if err != nil {
			logp.Err("Fail to set _ttl mapping: %s", err)
			// keep trying in the background
			go func() {
				for {
					err := out.EnableTTL()
					if err == nil {
						break
					}
					logp.Err("Fail to set _ttl mapping: %s", err)
					time.Sleep(5 * time.Second)
				}
			}()
		}
	}

	out.sendingQueue = make(chan EventMsg, 1000)
	go out.SendMessagesGoroutine()

	return nil
}

func (l *StdinInput) Init(config inputs.MothershipConfig) error {
	l.Type = "stdin"
	l.Config = config
	logp.Info("[StdinInput] Initialized")
	return nil
}

func (l *PackagesInput) Init(config inputs.MothershipConfig) error {
	l.Config = config
	l.Type = "Packages"
	logp.Info("[PackagesInput] Initialized")
	return nil
}

func (l *RedisInput) readKey(server redis.Conn, key string) ([]common.MapStr, uint64, int64, string, error) {
	var offset uint64 = 0
	var line uint64 = 0
	var prevTime uint64 = 0
	var thisTime uint64 = 0

	var events []common.MapStr

	var popScript = redis.NewScript(1, `return redis.call('LPOP', KEYS[1])`)
	var pushScript = redis.NewScript(2, `return redis.call('LPUSH', KEYS[1], KEYS[2])`)

	logp.Debug("redisinput", "Reading events from %s", key)

	for {
		reply, err := popScript.Do(server, key)
		if err != nil {
			logp.Info("[RedisInput] Unexpected state reading from %s; error: %s\n", key, err)
			return nil, line, 0, "", err
		}

		if reply == nil {
			logp.Debug("redisinputlines", "No values to read in LIST: %s", key)
			return events, line, int64(thisTime), "", nil
		}

		text, err := redis.String(reply, err)
		if err != nil {
			logp.Info("[RedisInput] Unexpected state converting reply to String; error: %s\n", err)
			return nil, line, 0, "", err
		}
		offset += uint64(len(text))
		line++

		event := common.MapStr{}
		event["source"] = strings.TrimSpace(key)
		event["offset"] = offset
		event["line"] = line
		event["message"] = &text
		event["type"] = strings.TrimSpace(l.Type)
		expanded_event, err := l.Filter(event)

		if _, present := expanded_event["metric_timestamp"]; present == false {
			if _, present := expanded_event["timestamp"]; present == false {
				logp.Err("This event has no timestamp field: %v\n", event)
				continue
			}
			expanded_event["metric_timestamp"] = expanded_event["timestamp"]
		}

		if _, present := expanded_event["metric_name"]; present == false {
			logp.Err("No metric_name found for: %v", event)
			continue
		}

		if _, present := expanded_event["metric_value"]; present == false {
			logp.Err("No metric_value found for: %v", event)
			continue
		}

		metricTime, err := strconv.ParseInt(expanded_event["metric_timestamp"].(string), 10, 64)
		if err != nil {
			logp.Err("An error parsing the metric_timestamp: %s\n", err)
		}
		thisTime = uint64(metricTime)

		_, nowMin, _ := time.Now().Clock()

		prevTime_Time := time.Unix(int64(prevTime), 0)
		_, prevMin, _ := prevTime_Time.Clock()

		thisTime_Time := time.Unix(int64(thisTime), 0)
		_, thisMin, _ := thisTime_Time.Clock()

		event["timestamp"] = thisTime_Time.Format("2006-01-02T15:04:05Z07:00")

		logp.Debug("timestuff", "This Minute: %v, Prev Minute: %v, Now Minute: %v", thisMin, prevMin, nowMin)

		// If it has not been a minute since this event happened, put it back in the list.
		// TODO: change this to see if event is older than 60 seconds
		if nowMin == thisMin {
			logp.Debug("redisinput", "Skipping, not old enough")
			logp.Debug("timestuff", "pushing event: this min is still the current min")
			pushScript.Do(server, key, text)
			if len(events) > 0 {
				logp.Debug("timestuff", "returning previously collected events")
				events, err := l.GroupEvents(events)
				if err != nil {
					logp.Err("An error occured while grouping the events: %v\n", err)
				}
				return events, line, int64(thisTime), expanded_event["metric_tags"].(string), nil
			} else {
				logp.Debug("timestuff", "sleeping 5 seconds, no collected events yet")
				time.Sleep(5 * time.Second)
			}
		} else {
			if thisMin <= prevMin || prevMin == 0 {
				prevTime = thisTime
				logp.Debug("timestuff", "appending event: this min is older than prev min, or prev min is 0")
				events = append(events, expanded_event)
			} else {
//.........这里部分代码省略.........