func addIdentifiers(csvFilename string, dbMap *gorp.DbMap, stats statsd.Statter, statsRate float32) {
	file, err := os.Open(csvFilename)
	cmd.FailOnError(err, "Could not open the file for reading")
	csvReader := csv.NewReader(file)

	for {
		record, err := csvReader.Read()
		if err == io.EOF {
			break
		} else if err != nil {
			fmt.Println("Error:", err)
			return
		}

		identifierData := core.IdentifierData{
			ReversedName: record[1],
			CertSHA1:     record[0],
		}

		importStart := time.Now()
		err = dbMap.Insert(&identifierData)
		stats.TimingDuration("ExistingCert.DomainImportTime", time.Since(importStart), statsRate)
		stats.Inc("ExistingCert.DomainsImported", 1, statsRate)
	}
}

func addCerts(csvFilename string, dbMap *gorp.DbMap, stats statsd.Statter, statsRate float32) {
	file, err := os.Open(csvFilename)
	cmd.FailOnError(err, "Could not open the file for reading")
	csvReader := csv.NewReader(file)

	for {
		record, err := csvReader.Read()
		if err == io.EOF {
			break
		} else if err != nil {
			fmt.Println("Error:", err)
			return
		}

		notAfter, err := time.Parse(datestampFormat, record[3])
		spkiBytes, err := hex.DecodeString(record[4])
		certDER, err := hex.DecodeString(record[7])

		externalCert := core.ExternalCert{
			SHA1:     record[0],
			Issuer:   record[1],
			Subject:  record[2],
			NotAfter: notAfter,
			SPKI:     spkiBytes,
			Valid:    record[5] == "1",
			EV:       record[6] == "1",
			CertDER:  certDER,
		}

		importStart := time.Now()
		err = dbMap.Insert(&externalCert)
		stats.TimingDuration("ExistingCert.CertImportTime", time.Since(importStart), statsRate)
		stats.Inc("ExistingCert.CertsImported", 1, statsRate)
	}
}

func validateContacts(contacts []*core.AcmeURL, resolver core.DNSResolver, stats statsd.Statter) (err error) {
	for _, contact := range contacts {
		switch contact.Scheme {
		case "tel":
			continue
		case "mailto":
			rtt, err := validateEmail(contact.Opaque, resolver)
			stats.TimingDuration("RA.DNS.RTT.MX", rtt, 1.0)
			stats.Inc("RA.DNS.Rate", 1, 1.0)
			if err != nil {
				return err
			}
		default:
			err = core.MalformedRequestError(fmt.Sprintf("Contact method %s is not supported", contact.Scheme))
			return
		}
	}

	return
}

func removeInvalidCerts(csvFilename string, dbMap *gorp.DbMap, stats statsd.Statter, statsRate float32) {
	file, err := os.Open(csvFilename)
	cmd.FailOnError(err, "Could not open the file for reading")
	csvReader := csv.NewReader(file)

	for {
		record, err := csvReader.Read()
		if err == io.EOF {
			break
		} else if err != nil {
			fmt.Println("Error:", err)
			return
		}

		identifierData := core.IdentifierData{
			CertSHA1: record[0],
		}
		externalCert := core.ExternalCert{
			SHA1: record[0],
		}

		deleteStart := time.Now()
		_, err = dbMap.Delete(&identifierData)
		stats.TimingDuration("ExistingCert.DomainDeleteTime", time.Since(deleteStart), statsRate)
		_, err = dbMap.Delete(&externalCert)
		stats.TimingDuration("ExistingCert.CertDeleteTime", time.Since(deleteStart), statsRate)
		stats.Inc("ExistingCert.CertsDeleted", 1, statsRate)
	}
}

func timeDelivery(d amqp.Delivery, stats statsd.Statter, deliveryTimings map[string]time.Time) {
	// If d is a call add to deliveryTimings and increment openCalls, if it is a
	// response then get time.Since original call from deliveryTiming, send timing metric, and
	// decrement openCalls, in both cases send the gauges RpcCallsOpen and RpcBodySize
	if d.ReplyTo != "" {
		openCalls++
		deliveryTimings[fmt.Sprintf("%s:%s", d.CorrelationId, d.ReplyTo)] = time.Now()
	} else {
		openCalls--
		rpcSent := deliveryTimings[fmt.Sprintf("%s:%s", d.CorrelationId, d.RoutingKey)]
		if rpcSent != *new(time.Time) {
			respTime := time.Since(rpcSent)
			delete(deliveryTimings, fmt.Sprintf("%s:%s", d.CorrelationId, d.RoutingKey))

			stats.TimingDuration(fmt.Sprintf("RpcCallTime.%s", d.Type), respTime, 1.0)
		}
	}

	stats.Gauge("RpcCallsOpen", openCalls, 1.0)
	stats.Gauge("RpcBodySize", int64(len(d.Body)), 1.0)
}

// HandlerTimer monitors HTTP performance and sends the details to StatsD.
func HandlerTimer(handler http.Handler, stats statsd.Statter) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		cStart := time.Now()
		openConnections++
		stats.Gauge("HttpConnectionsOpen", openConnections, 1.0)

		handler.ServeHTTP(w, r)

		openConnections--
		stats.Gauge("HttpConnectionsOpen", openConnections, 1.0)

		// (FIX: this doesn't seem to really work at catching errors...)
		state := "Success"
		for _, h := range w.Header()["Content-Type"] {
			if h == "application/problem+json" {
				state = "Error"
				break
			}
		}
		// set resp timing key based on success / failure
		stats.TimingDuration(fmt.Sprintf("HttpResponseTime.%s.%s", r.URL, state), time.Since(cStart), 1.0)
	})
}

// ProfileCmd runs forever, sending Go statistics to StatsD.
func ProfileCmd(profileName string, stats statsd.Statter) {
	for {
		var memoryStats runtime.MemStats
		runtime.ReadMemStats(&memoryStats)

		stats.Gauge(fmt.Sprintf("Gostats.%s.Goroutines", profileName), int64(runtime.NumGoroutine()), 1.0)

		stats.Gauge(fmt.Sprintf("Gostats.%s.Heap.Objects", profileName), int64(memoryStats.HeapObjects), 1.0)
		stats.Gauge(fmt.Sprintf("Gostats.%s.Heap.Idle", profileName), int64(memoryStats.HeapIdle), 1.0)
		stats.Gauge(fmt.Sprintf("Gostats.%s.Heap.InUse", profileName), int64(memoryStats.HeapInuse), 1.0)
		stats.Gauge(fmt.Sprintf("Gostats.%s.Heap.Released", profileName), int64(memoryStats.HeapReleased), 1.0)

		gcPauseAvg := int64(memoryStats.PauseTotalNs) / int64(len(memoryStats.PauseNs))

		stats.Timing(fmt.Sprintf("Gostats.%s.Gc.PauseAvg", profileName), gcPauseAvg, 1.0)
		stats.Gauge(fmt.Sprintf("Gostats.%s.Gc.NextAt", profileName), int64(memoryStats.NextGC), 1.0)

		time.Sleep(time.Second)
	}
}

// ProfileCmd runs forever, sending Go runtime statistics to StatsD.
func ProfileCmd(profileName string, stats statsd.Statter) {
	var memoryStats runtime.MemStats
	prevNumGC := int64(0)
	c := time.Tick(1 * time.Second)
	for range c {
		runtime.ReadMemStats(&memoryStats)

		// Gather goroutine count
		stats.Gauge(fmt.Sprintf("%s.Gostats.Goroutines", profileName), int64(runtime.NumGoroutine()), 1.0)

		// Gather various heap metrics
		stats.Gauge(fmt.Sprintf("%s.Gostats.Heap.Alloc", profileName), int64(memoryStats.HeapAlloc), 1.0)
		stats.Gauge(fmt.Sprintf("%s.Gostats.Heap.Objects", profileName), int64(memoryStats.HeapObjects), 1.0)
		stats.Gauge(fmt.Sprintf("%s.Gostats.Heap.Idle", profileName), int64(memoryStats.HeapIdle), 1.0)
		stats.Gauge(fmt.Sprintf("%s.Gostats.Heap.InUse", profileName), int64(memoryStats.HeapInuse), 1.0)
		stats.Gauge(fmt.Sprintf("%s.Gostats.Heap.Released", profileName), int64(memoryStats.HeapReleased), 1.0)

		// Gather various GC related metrics
		if memoryStats.NumGC > 0 {
			totalRecentGC := uint64(0)
			realBufSize := uint32(256)
			if memoryStats.NumGC < 256 {
				realBufSize = memoryStats.NumGC
			}
			for _, pause := range memoryStats.PauseNs {
				totalRecentGC += pause
			}
			gcPauseAvg := totalRecentGC / uint64(realBufSize)
			lastGC := memoryStats.PauseNs[(memoryStats.NumGC+255)%256]
			stats.Timing(fmt.Sprintf("%s.Gostats.Gc.PauseAvg", profileName), int64(gcPauseAvg), 1.0)
			stats.Gauge(fmt.Sprintf("%s.Gostats.Gc.LastPause", profileName), int64(lastGC), 1.0)
		}
		stats.Gauge(fmt.Sprintf("%s.Gostats.Gc.NextAt", profileName), int64(memoryStats.NextGC), 1.0)
		// Send both a counter and a gauge here we can much more easily observe
		// the GC rate (versus the raw number of GCs) in graphing tools that don't
		// like deltas
		stats.Gauge(fmt.Sprintf("%s.Gostats.Gc.Count", profileName), int64(memoryStats.NumGC), 1.0)
		gcInc := int64(memoryStats.NumGC) - prevNumGC
		stats.Inc(fmt.Sprintf("%s.Gostats.Gc.Rate", profileName), gcInc, 1.0)
		prevNumGC += gcInc
	}
}

// NewAmqpRPCClient constructs an RPC client using AMQP
func NewAmqpRPCClient(clientQueuePrefix, serverQueue string, c cmd.Config, stats statsd.Statter) (rpc *AmqpRPCCLient, err error) {
	hostname, err := os.Hostname()
	if err != nil {
		return nil, err
	}

	randID := make([]byte, 3)
	_, err = rand.Read(randID)
	if err != nil {
		return nil, err
	}
	clientQueue := fmt.Sprintf("%s.%s.%x", clientQueuePrefix, hostname, randID)

	reconnectBase := c.AMQP.ReconnectTimeouts.Base.Duration
	if reconnectBase == 0 {
		reconnectBase = 20 * time.Millisecond
	}
	reconnectMax := c.AMQP.ReconnectTimeouts.Max.Duration
	if reconnectMax == 0 {
		reconnectMax = time.Minute
	}

	rpc = &AmqpRPCCLient{
		serverQueue: serverQueue,
		clientQueue: clientQueue,
		connection:  newAMQPConnector(clientQueue, reconnectBase, reconnectMax),
		pending:     make(map[string]chan []byte),
		timeout:     10 * time.Second,
		log:         blog.GetAuditLogger(),
		stats:       stats,
	}

	err = rpc.connection.connect(c)
	if err != nil {
		return nil, err
	}

	go func() {
		for {
			select {
			case msg, ok := <-rpc.connection.messages():
				if ok {
					corrID := msg.CorrelationId
					rpc.mu.RLock()
					responseChan, present := rpc.pending[corrID]
					rpc.mu.RUnlock()

					if !present {
						// occurs when a request is timed out and the arrives
						// afterwards
						stats.Inc("RPC.AfterTimeoutResponseArrivals."+clientQueuePrefix, 1, 1.0)
						continue
					}

					responseChan <- msg.Body
					rpc.mu.Lock()
					delete(rpc.pending, corrID)
					rpc.mu.Unlock()
				} else {
					// chan has been closed by rpc.connection.Cancel
					rpc.log.Info(fmt.Sprintf(" [!] Client reply channel closed: %s", rpc.clientQueue))
					continue
				}
			case err = <-rpc.connection.closeChannel():
				rpc.log.Info(fmt.Sprintf(" [!] Client reply channel closed : %s", rpc.clientQueue))
				rpc.connection.reconnect(c, rpc.log)
			}
		}
	}()

	return rpc, err
}

func main() {

	// command line flags
	var opts struct {
		HostPort  string        `long:"host" default:"127.0.0.1:8125" description:"host:port of statsd server"`
		Prefix    string        `long:"prefix" default:"test-client" description:"Statsd prefix"`
		StatType  string        `long:"type" default:"count" description:"stat type to send. Can be timing, count, guage"`
		StatValue int64         `long:"value" default:"1" description:"Value to send"`
		Name      string        `short:"n" long:"name" default:"counter" description:"stat name"`
		Rate      float32       `short:"r" long:"rate" default:"1.0" description:"sample rate"`
		Volume    int           `short:"c" long:"count" default:"1000" description:"Number of stats to send. Volume."`
		Nil       bool          `long:"nil" default:"false" description:"Use nil client"`
		Buffered  bool          `long:"buffered" default:"false" description:"Use a buffered client"`
		Duration  time.Duration `short:"d" long:"duration" default:"10s" description:"How long to spread the volume across. Each second of duration volume/seconds events will be sent."`
	}

	// parse said flags
	_, err := flags.Parse(&opts)
	if err != nil {
		if e, ok := err.(*flags.Error); ok {
			if e.Type == flags.ErrHelp {
				os.Exit(0)
			}
		}
		fmt.Printf("Error: %+v\n", err)
		os.Exit(1)
	}

	if opts.Nil && opts.Buffered {
		fmt.Printf("Specifying both nil and buffered together is invalid.")
		os.Exit(1)
	}

	var client statsd.Statter
	if !opts.Nil {
		if !opts.Buffered {
			client, err = statsd.NewClient(opts.HostPort, opts.Prefix)
		} else {
			client, err = statsd.NewBufferedClient(opts.HostPort, opts.Prefix, opts.Duration/time.Duration(4), 0)
		}
		if err != nil {
			log.Fatal(err)
		}
		defer client.Close()
	}

	var stat func(stat string, value int64, rate float32) error
	switch opts.StatType {
	case "count":
		stat = func(stat string, value int64, rate float32) error {
			return client.Inc(stat, value, rate)
		}
	case "gauge":
		stat = func(stat string, value int64, rate float32) error {
			return client.Gauge(stat, value, rate)
		}
	case "timing":
		stat = func(stat string, value int64, rate float32) error {
			return client.Timing(stat, value, rate)
		}
	default:
		log.Fatal("Unsupported state type")
	}

	pertick := opts.Volume / int(opts.Duration.Seconds()) / 10
	// add some extra tiem, because the first tick takes a while
	ender := time.After(opts.Duration + 100*time.Millisecond)
	c := time.Tick(time.Second / 10)
	count := 0
	for {
		select {
		case <-c:
			for x := 0; x < pertick; x++ {
				err := stat(opts.Name, opts.StatValue, opts.Rate)
				if err != nil {
					log.Printf("Got Error: %+v", err)
					break
				}
				count += 1
			}
		case <-ender:
			log.Printf("%d events called", count)
			os.Exit(0)
			return
		}
	}
}

// ProfileCmd runs forever, sending Go runtime statistics to StatsD.
func ProfileCmd(profileName string, stats statsd.Statter) {
	c := time.Tick(1 * time.Second)
	for range c {
		var memoryStats runtime.MemStats
		runtime.ReadMemStats(&memoryStats)

		stats.Gauge(fmt.Sprintf("%s.Gostats.Goroutines", profileName), int64(runtime.NumGoroutine()), 1.0)

		stats.Gauge(fmt.Sprintf("%s.Gostats.Heap.Alloc", profileName), int64(memoryStats.HeapAlloc), 1.0)
		stats.Gauge(fmt.Sprintf("%s.Gostats.Heap.Objects", profileName), int64(memoryStats.HeapObjects), 1.0)
		stats.Gauge(fmt.Sprintf("%s.Gostats.Heap.Idle", profileName), int64(memoryStats.HeapIdle), 1.0)
		stats.Gauge(fmt.Sprintf("%s.Gostats.Heap.InUse", profileName), int64(memoryStats.HeapInuse), 1.0)
		stats.Gauge(fmt.Sprintf("%s.Gostats.Heap.Released", profileName), int64(memoryStats.HeapReleased), 1.0)

		// Calculate average and last and convert from nanoseconds to milliseconds
		gcPauseAvg := (int64(memoryStats.PauseTotalNs) / int64(len(memoryStats.PauseNs))) / 1000000
		lastGC := int64(memoryStats.PauseNs[(memoryStats.NumGC+255)%256]) / 1000000
		stats.Timing(fmt.Sprintf("%s.Gostats.Gc.PauseAvg", profileName), gcPauseAvg, 1.0)
		stats.Gauge(fmt.Sprintf("%s.Gostats.Gc.LastPauseLatency", profileName), lastGC, 1.0)
		stats.Gauge(fmt.Sprintf("%s.Gostats.Gc.NextAt", profileName), int64(memoryStats.NextGC), 1.0)
	}
}