Repository: VictoriaMetrics/VictoriaMetrics
Pull request #332 (2020-02-25 11:37:50)

Readme envvars
As discussed in #311 here is a proposal about readme/env vars. Feel free to ask modifications. Once settled, I will update and PR the cluster readme as well.


Patch:
diff --git a/README.md b/README.md
index dc63a5597..4b8410322 100644
--- a/README.md
+++ b/README.md
@@ -57,7 +57,7 @@ Cluster version is available [here](https://github.com/VictoriaMetrics/VictoriaM
   * [Metrics from Prometheus exporters](https://github.com/prometheus/docs/blob/master/content/docs/instrumenting/exposition_formats.md#text-based-format)
   such as [node_exporter](https://github.com/prometheus/node_exporter). See [these docs](#how-to-scrape-prometheus-exporters-such-as-node-exporter) for details.
   * [Prometheus remote write API](https://prometheus.io/docs/prometheus/latest/configuration/configuration/#remote_write)
-  * [InfluxDB line protocol](#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf)
+  * [InfluxDB line protocol](#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf) over HTTP, TCP and UDP.
   * [Graphite plaintext protocol](#how-to-send-data-from-graphite-compatible-agents-such-as-statsd) with [tags](https://graphite.readthedocs.io/en/latest/tags.html#carbon)
     if `-graphiteListenAddr` is set.
   * [OpenTSDB put message](#sending-data-via-telnet-put-protocol) if `-opentsdbListenAddr` is set.
@@ -73,6 +73,7 @@ Cluster version is available [here](https://github.com/VictoriaMetrics/VictoriaM
 ### Table of contents
 
   - [How to start VictoriaMetrics](#how-to-start-victoriametrics)
+      - [Environment variables](#environment-variables)
   - [Prometheus setup](#prometheus-setup)
   - [Grafana setup](#grafana-setup)
   - [How to upgrade VictoriaMetrics?](#how-to-upgrade-victoriametrics)
@@ -137,12 +138,15 @@ The following command-line flags are used the most:
 
 Pass `-help` to see all the available flags with description and default values.
 
-Default flag values may be read from environment variables if `-envflag.enable` command-line flag is set.
-Substitute dots with underscores in env var names. Alternative syntax can be used for setting repeatable flags:
-`-arg=foo -arg=bar` can be written as `-arg=foo,bar`. See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/311) for more details.
-
 It is recommended setting up [monitoring](#monitoring) for VictoriaMetrics.
 
+#### Environment variables
+
+Each flag values can be set thru environment variables by following these rules:
+
+* The `-envflag.enable` flag must be set
+* Each `.` in flag names must be substituted by `_` (for example `-insert.maxQueueDuration <duration>` will translate to `insert_maxQueueDuration=<duration>`)
+* For repeating flags, an alternative syntax can be used by joining the different values into one using `,` as separator (for example `-storageNode <nodeA> -storageNode <nodeB>` will translate to `storageNode=<nodeA>,<nodeB>`)
 
 ### Prometheus setup
 
@@ -268,6 +272,8 @@ For instance, put the following lines into `Telegraf` config, so it sends data t
 
 Do not forget substituting `<victoriametrics-addr>` with the real address where VictoriaMetrics runs.
 
+Another option is to enable TCP and UDP receiver for Influx line protocol via `-influxListenAddr` command-line flag.
+
 VictoriaMetrics maps Influx data using the following rules:
 * [`db` query arg](https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint) is mapped into `db` label value
   unless `db` tag exists in the Influx line.
diff --git a/app/vmagent/README.md b/app/vmagent/README.md
index 20781cc74..432a902cd 100644
--- a/app/vmagent/README.md
+++ b/app/vmagent/README.md
@@ -118,9 +118,11 @@ Read more about relabeling in the following articles:
 
 ### Monitoring
 
-`vmagent` exports various metrics in Prometheus exposition format at `/metrics` page. It is recommended setting up regular scraping of this page
+`vmagent` exports various metrics in Prometheus exposition format at `http://vmagent-host:8429/metrics` page. It is recommended setting up regular scraping of this page
 either via `vmagent` itself or via Prometheus, so the exported metrics could be analyzed later.
 
+`vmagent` also exports target statuses at `http://vmagent-host:8429/targets` page in plaintext format.
+
 
 ### Troubleshooting
 
diff --git a/app/vmagent/influx/request_handler.go b/app/vmagent/influx/request_handler.go
index e0ed2ad92..642bd3d87 100644
--- a/app/vmagent/influx/request_handler.go
+++ b/app/vmagent/influx/request_handler.go
@@ -2,6 +2,7 @@ package influx
 
 import (
 	"flag"
+	"io"
 	"net/http"
 	"runtime"
 	"sync"
@@ -25,12 +26,26 @@ var (
 	rowsPerInsert = metrics.NewHistogram(`vmagent_rows_per_insert{type="influx"}`)
 )
 
-// InsertHandler processes remote write for influx line protocol.
+// InsertHandlerForReader processes remote write for influx line protocol.
+//
+// See https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener/
+func InsertHandlerForReader(r io.Reader) error {
+	return writeconcurrencylimiter.Do(func() error {
+		return parser.ParseStream(r, false, "", "", insertRows)
+	})
+}
+
+// InsertHandlerForHTTP processes remote write for influx line protocol.
 //
 // See https://github.com/influxdata/influxdb/blob/4cbdc197b8117fee648d62e2e5be75c6575352f0/tsdb/README.md
-func InsertHandler(req *http.Request) error {
+func InsertHandlerForHTTP(req *http.Request) error {
 	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(req, insertRows)
+		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+		q := req.URL.Query()
+		precision := q.Get("precision")
+		// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
+		db := q.Get("db")
+		return parser.ParseStream(req.Body, isGzipped, precision, db, insertRows)
 	})
 }
 
diff --git a/app/vmagent/main.go b/app/vmagent/main.go
index 264a0f5a9..79a32e048 100644
--- a/app/vmagent/main.go
+++ b/app/vmagent/main.go
@@ -18,6 +18,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	graphiteserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/graphite"
+	influxserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/influx"
 	opentsdbserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdb"
 	opentsdbhttpserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdbhttp"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
@@ -29,6 +30,7 @@ import (
 
 var (
 	httpListenAddr     = flag.String("httpListenAddr", ":8429", "TCP address to listen for http connections")
+	influxListenAddr   = flag.String("influxListenAddr", "", "TCP and UDP address to listen for Influx line protocol data. Usually :8189 must be set. Doesn't work if empty")
 	graphiteListenAddr = flag.String("graphiteListenAddr", "", "TCP and UDP address to listen for Graphite plaintext data. Usually :2003 must be set. Doesn't work if empty")
 	opentsdbListenAddr = flag.String("opentsdbListenAddr", "", "TCP and UDP address to listen for OpentTSDB metrics. "+
 		"Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. "+
@@ -37,6 +39,7 @@ var (
 )
 
 var (
+	influxServer       *influxserver.Server
 	graphiteServer     *graphiteserver.Server
 	opentsdbServer     *opentsdbserver.Server
 	opentsdbhttpServer *opentsdbhttpserver.Server
@@ -50,6 +53,9 @@ func main() {
 	startTime := time.Now()
 	remotewrite.Init()
 	writeconcurrencylimiter.Init()
+	if len(*influxListenAddr) > 0 {
+		influxServer = influxserver.MustStart(*influxListenAddr, influx.InsertHandlerForReader)
+	}
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer = graphiteserver.MustStart(*graphiteListenAddr, graphite.InsertHandler)
 	}
@@ -77,6 +83,9 @@ func main() {
 
 	promscrape.Stop()
 
+	if len(*influxListenAddr) > 0 {
+		influxServer.MustStop()
+	}
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer.MustStop()
 	}
@@ -103,10 +112,6 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		}
 		w.WriteHeader(http.StatusNoContent)
 		return true
-	case "/targets":
-		w.Header().Set("Content-Type", "text/plain")
-		promscrape.WriteHumanReadableTargetsStatus(w)
-		return true
 	case "/api/v1/import":
 		vmimportRequests.Inc()
 		if err := vmimport.InsertHandler(r); err != nil {
@@ -118,7 +123,7 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	case "/write", "/api/v2/write":
 		influxWriteRequests.Inc()
-		if err := influx.InsertHandler(r); err != nil {
+		if err := influx.InsertHandlerForHTTP(r); err != nil {
 			influxWriteErrors.Inc()
 			httpserver.Errorf(w, "error in %q: %s", r.URL.Path, err)
 			return true
@@ -131,6 +136,11 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		influxQueryRequests.Inc()
 		fmt.Fprintf(w, `{"results":[{"series":[{"values":[]}]}]}`)
 		return true
+	case "/targets":
+		promscrapeTargetsRequests.Inc()
+		w.Header().Set("Content-Type", "text/plain")
+		promscrape.WriteHumanReadableTargetsStatus(w)
+		return true
 	}
 	return false
 }
@@ -146,4 +156,6 @@ var (
 	influxWriteErrors   = metrics.NewCounter(`vmagent_http_request_errors_total{path="/write", protocol="influx"}`)
 
 	influxQueryRequests = metrics.NewCounter(`vmagent_http_requests_total{path="/query", protocol="influx"}`)
+
+	promscrapeTargetsRequests = metrics.NewCounter(`vmagent_http_requests_total{path="/targets"}`)
 )
diff --git a/app/vmagent/remotewrite/pendingseries.go b/app/vmagent/remotewrite/pendingseries.go
index b583ebffa..3b64d3212 100644
--- a/app/vmagent/remotewrite/pendingseries.go
+++ b/app/vmagent/remotewrite/pendingseries.go
@@ -13,8 +13,12 @@ import (
 	"github.com/golang/snappy"
 )
 
-var flushInterval = flag.Duration("remoteWrite.flushInterval", time.Second, "Interval for flushing the data to remote storage. "+
-	"Higher value reduces network bandwidth usage at the cost of delayed push of scraped data to remote storage")
+var (
+	flushInterval = flag.Duration("remoteWrite.flushInterval", time.Second, "Interval for flushing the data to remote storage. "+
+		"Higher value reduces network bandwidth usage at the cost of delayed push of scraped data to remote storage")
+	maxUnpackedBlockSize = flag.Int("remoteWrite.maxBlockSize", 32*1024*1024, "The maximum size in bytes of unpacked request to send to remote storage. "+
+		"It shouldn't exceed -maxInsertRequestSize from VictoriaMetrics")
+)
 
 // the maximum number of rows to send per each block.
 const maxRowsPerBlock = 10000
@@ -160,17 +164,21 @@ func pushWriteRequest(wr *prompbmarshal.WriteRequest, pushBlock func(block []byt
 	}
 	bb := writeRequestBufPool.Get()
 	bb.B = prompbmarshal.MarshalWriteRequest(bb.B[:0], wr)
-	zb := snappyBufPool.Get()
-	zb.B = snappy.Encode(zb.B[:cap(zb.B)], bb.B)
-	writeRequestBufPool.Put(bb)
-	if len(zb.B) <= persistentqueue.MaxBlockSize {
-		pushBlock(zb.B)
-		blockSizeRows.Update(float64(len(wr.Timeseries)))
-		blockSizeBytes.Update(float64(len(zb.B)))
+	if len(bb.B) <= *maxUnpackedBlockSize {
+		zb := snappyBufPool.Get()
+		zb.B = snappy.Encode(zb.B[:cap(zb.B)], bb.B)
+		writeRequestBufPool.Put(bb)
+		if len(zb.B) <= persistentqueue.MaxBlockSize {
+			pushBlock(zb.B)
+			blockSizeRows.Update(float64(len(wr.Timeseries)))
+			blockSizeBytes.Update(float64(len(zb.B)))
+			snappyBufPool.Put(zb)
+			return
+		}
 		snappyBufPool.Put(zb)
-		return
+	} else {
+		writeRequestBufPool.Put(bb)
 	}
-	snappyBufPool.Put(zb)
 
 	// Too big block. Recursively split it into smaller parts.
 	timeseries := wr.Timeseries
diff --git a/app/vminsert/influx/request_handler.go b/app/vminsert/influx/request_handler.go
index 43d03457e..cb25d1205 100644
--- a/app/vminsert/influx/request_handler.go
+++ b/app/vminsert/influx/request_handler.go
@@ -2,6 +2,7 @@ package influx
 
 import (
 	"flag"
+	"io"
 	"net/http"
 	"runtime"
 	"sync"
@@ -24,12 +25,26 @@ var (
 	rowsPerInsert = metrics.NewHistogram(`vm_rows_per_insert{type="influx"}`)
 )
 
-// InsertHandler processes remote write for influx line protocol.
+// InsertHandlerForReader processes remote write for influx line protocol.
+//
+// See https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener/
+func InsertHandlerForReader(r io.Reader) error {
+	return writeconcurrencylimiter.Do(func() error {
+		return parser.ParseStream(r, false, "", "", insertRows)
+	})
+}
+
+// InsertHandlerForHTTP processes remote write for influx line protocol.
 //
 // See https://github.com/influxdata/influxdb/blob/4cbdc197b8117fee648d62e2e5be75c6575352f0/tsdb/README.md
-func InsertHandler(req *http.Request) error {
+func InsertHandlerForHTTP(req *http.Request) error {
 	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(req, insertRows)
+		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+		q := req.URL.Query()
+		precision := q.Get("precision")
+		// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
+		db := q.Get("db")
+		return parser.ParseStream(req.Body, isGzipped, precision, db, insertRows)
 	})
 }
 
diff --git a/app/vminsert/main.go b/app/vminsert/main.go
index 79dd90d4c..80c4359f2 100644
--- a/app/vminsert/main.go
+++ b/app/vminsert/main.go
@@ -15,6 +15,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/vmimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	graphiteserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/graphite"
+	influxserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/influx"
 	opentsdbserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdb"
 	opentsdbhttpserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdbhttp"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
@@ -25,6 +26,7 @@ import (
 
 var (
 	graphiteListenAddr = flag.String("graphiteListenAddr", "", "TCP and UDP address to listen for Graphite plaintext data. Usually :2003 must be set. Doesn't work if empty")
+	influxListenAddr   = flag.String("influxListenAddr", "", "TCP and UDP address to listen for Influx line protocol data. Usually :8189 must be set. Doesn't work if empty")
 	opentsdbListenAddr = flag.String("opentsdbListenAddr", "", "TCP and UDP address to listen for OpentTSDB metrics. "+
 		"Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. "+
 		"Usually :4242 must be set. Doesn't work if empty")
@@ -33,6 +35,7 @@ var (
 )
 
 var (
+	influxServer       *influxserver.Server
 	graphiteServer     *graphiteserver.Server
 	opentsdbServer     *opentsdbserver.Server
 	opentsdbhttpServer *opentsdbhttpserver.Server
@@ -43,6 +46,9 @@ func Init() {
 	storage.SetMaxLabelsPerTimeseries(*maxLabelsPerTimeseries)
 
 	writeconcurrencylimiter.Init()
+	if len(*influxListenAddr) > 0 {
+		influxServer = influxserver.MustStart(*influxListenAddr, influx.InsertHandlerForReader)
+	}
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer = graphiteserver.MustStart(*graphiteListenAddr, graphite.InsertHandler)
 	}
@@ -58,6 +64,9 @@ func Init() {
 // Stop stops vminsert.
 func Stop() {
 	promscrape.Stop()
+	if len(*influxListenAddr) > 0 {
+		influxServer.MustStop()
+	}
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer.MustStop()
 	}
@@ -93,7 +102,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	case "/write", "/api/v2/write":
 		influxWriteRequests.Inc()
-		if err := influx.InsertHandler(r); err != nil {
+		if err := influx.InsertHandlerForHTTP(r); err != nil {
 			influxWriteErrors.Inc()
 			httpserver.Errorf(w, "error in %q: %s", r.URL.Path, err)
 			return true
@@ -106,6 +115,11 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		influxQueryRequests.Inc()
 		fmt.Fprintf(w, `{"results":[{"series":[{"values":[]}]}]}`)
 		return true
+	case "/targets":
+		promscrapeTargetsRequests.Inc()
+		w.Header().Set("Content-Type", "text/plain")
+		promscrape.WriteHumanReadableTargetsStatus(w)
+		return true
 	default:
 		// This is not our link
 		return false
@@ -123,4 +137,6 @@ var (
 	influxWriteErrors   = metrics.NewCounter(`vm_http_request_errors_total{path="/write", protocol="influx"}`)
 
 	influxQueryRequests = metrics.NewCounter(`vm_http_requests_total{path="/query", protocol="influx"}`)
+
+	promscrapeTargetsRequests = metrics.NewCounter(`vm_http_requests_total{path="/targets"}`)
 )
diff --git a/lib/fs/reader_at.go b/lib/fs/reader_at.go
index 0acaedafb..26bf4df3f 100644
--- a/lib/fs/reader_at.go
+++ b/lib/fs/reader_at.go
@@ -32,7 +32,7 @@ func (r *ReaderAt) MustReadAt(p []byte, off int64) {
 	if len(p) == 0 {
 		return
 	}
-	if len(r.mmapData) == 0 || len(p) > 8*1024*1024 {
+	if len(r.mmapData) == 0 || len(p) > 8*1024 {
 		// Read big blocks directly from file.
 		// This could be faster than reading these blocks from mmap,
 		// since it triggers less page faults.
diff --git a/lib/ingestserver/graphite/server.go b/lib/ingestserver/graphite/server.go
index 41955a4c0..02be8da7b 100644
--- a/lib/ingestserver/graphite/server.go
+++ b/lib/ingestserver/graphite/server.go
@@ -15,11 +15,11 @@ import (
 )
 
 var (
-	writeRequestsTCP = metrics.NewCounter(`vm_ingestserver_graphite_requests_total{name="write", net="tcp"}`)
-	writeErrorsTCP   = metrics.NewCounter(`vm_ingestserver_graphite_request_errors_total{name="write", net="tcp"}`)
+	writeRequestsTCP = metrics.NewCounter(`vm_ingestserver_requests_total{type="graphite", name="write", net="tcp"}`)
+	writeErrorsTCP   = metrics.NewCounter(`vm_ingestserver_request_errors_total{type="graphite", name="write", net="tcp"}`)
 
-	writeRequestsUDP = metrics.NewCounter(`vm_ingestserver_graphite_requests_total{name="write", net="udp"}`)
-	writeErrorsUDP   = metrics.NewCounter(`vm_ingestserver_graphite_request_errors_total{name="write", net="udp"}`)
+	writeRequestsUDP = metrics.NewCounter(`vm_ingestserver_requests_total{type="graphite", name="write", net="udp"}`)
+	writeErrorsUDP   = metrics.NewCounter(`vm_ingestserver_request_errors_total{type="graphite", name="write", net="udp"}`)
 )
 
 // Server accepts Graphite plaintext lines over TCP and UDP.
diff --git a/lib/ingestserver/influx/server.go b/lib/ingestserver/influx/server.go
new file mode 100644
index 000000000..f429d9bfa
--- /dev/null
+++ b/lib/ingestserver/influx/server.go
@@ -0,0 +1,152 @@
+package influx
+
+import (
+	"io"
+	"net"
+	"runtime"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+var (
+	writeRequestsTCP = metrics.NewCounter(`vm_ingestserver_requests_total{type="influx", name="write", net="tcp"}`)
+	writeErrorsTCP   = metrics.NewCounter(`vm_ingestserver_request_errors_total{type="influx", name="write", net="tcp"}`)
+
+	writeRequestsUDP = metrics.NewCounter(`vm_ingestserver_requests_total{type="influx", name="write", net="udp"}`)
+	writeErrorsUDP   = metrics.NewCounter(`vm_ingestserver_request_errors_total{type="influx", name="write", net="udp"}`)
+)
+
+// Server accepts Influx line protocol over TCP and UDP.
+type Server struct {
+	addr  string
+	lnTCP net.Listener
+	lnUDP net.PacketConn
+	wg    sync.WaitGroup
+}
+
+// MustStart starts Influx server on the given addr.
+//
+// The incoming connections are processed with insertHandler.
+//
+// MustStop must be called on the returned server when it is no longer needed.
+func MustStart(addr string, insertHandler func(r io.Reader) error) *Server {
+	logger.Infof("starting TCP Influx server at %q", addr)
+	lnTCP, err := netutil.NewTCPListener("influx", addr)
+	if err != nil {
+		logger.Fatalf("cannot start TCP Influx server at %q: %s", addr, err)
+	}
+
+	logger.Infof("starting UDP Influx server at %q", addr)
+	lnUDP, err := net.ListenPacket("udp4", addr)
+	if err != nil {
+		logger.Fatalf("cannot start UDP Influx server at %q: %s", addr, err)
+	}
+
+	s := &Server{
+		addr:  addr,
+		lnTCP: lnTCP,
+		lnUDP: lnUDP,
+	}
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+		serveTCP(lnTCP, insertHandler)
+		logger.Infof("stopped TCP Influx server at %q", addr)
+	}()
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+		serveUDP(lnUDP, insertHandler)
+		logger.Infof("stopped UDP Influx server at %q", addr)
+	}()
+	return s
+}
+
+// MustStop stops the server.
+func (s *Server) MustStop() {
+	logger.Infof("stopping TCP Influx server at %q...", s.addr)
+	if err := s.lnTCP.Close(); err != nil {
+		logger.Errorf("cannot close TCP Influx server: %s", err)
+	}
+	logger.Infof("stopping UDP Influx server at %q...", s.addr)
+	if err := s.lnUDP.Close(); err != nil {
+		logger.Errorf("cannot close UDP Influx server: %s", err)
+	}
+	s.wg.Wait()
+	logger.Infof("TCP and UDP Influx servers at %q have been stopped", s.addr)
+}
+
+func serveTCP(ln net.Listener, insertHandler func(r io.Reader) error) {
+	for {
+		c, err := ln.Accept()
+		if err != nil {
+			if ne, ok := err.(net.Error); ok {
+				if ne.Temporary() {
+					logger.Errorf("influx: temporary error when listening for TCP addr %q: %s", ln.Addr(), err)
+					time.Sleep(time.Second)
+					continue
+				}
+				if strings.Contains(err.Error(), "use of closed network connection") {
+					break
+				}
+				logger.Fatalf("unrecoverable error when accepting TCP Influx connections: %s", err)
+			}
+			logger.Fatalf("unexpected error when accepting TCP Influx connections: %s", err)
+		}
+		go func() {
+			writeRequestsTCP.Inc()
+			if err := insertHandler(c); err != nil {
+				writeErrorsTCP.Inc()
+				logger.Errorf("error in TCP Influx conn %q<->%q: %s", c.LocalAddr(), c.RemoteAddr(), err)
+			}
+			_ = c.Close()
+		}()
+	}
+}
+
+func serveUDP(ln net.PacketConn, insertHandler func(r io.Reader) error) {
+	gomaxprocs := runtime.GOMAXPROCS(-1)
+	var wg sync.WaitGroup
+	for i := 0; i < gomaxprocs; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			var bb bytesutil.ByteBuffer
+			bb.B = bytesutil.Resize(bb.B, 64*1024)
+			for {
+				bb.Reset()
+				bb.B = bb.B[:cap(bb.B)]
+				n, addr, err := ln.ReadFrom(bb.B)
+				if err != nil {
+					writeErrorsUDP.Inc()
+					if ne, ok := err.(net.Error); ok {
+						if ne.Temporary() {
+							logger.Errorf("influx: temporary error when listening for UDP addr %q: %s", ln.LocalAddr(), err)
+							time.Sleep(time.Second)
+							continue
+						}
+						if strings.Contains(err.Error(), "use of closed network connection") {
+							break
+						}
+					}
+					logger.Errorf("cannot read Influx UDP data: %s", err)
+					continue
+				}
+				bb.B = bb.B[:n]
+				writeRequestsUDP.Inc()
+				if err := insertHandler(bb.NewReader()); err != nil {
+					writeErrorsUDP.Inc()
+					logger.Errorf("error in UDP Influx conn %q<->%q: %s", ln.LocalAddr(), addr, err)
+					continue
+				}
+			}
+		}()
+	}
+	wg.Wait()
+}
diff --git a/lib/ingestserver/opentsdb/server.go b/lib/ingestserver/opentsdb/server.go
index b38e217fe..af83bc923 100644
--- a/lib/ingestserver/opentsdb/server.go
+++ b/lib/ingestserver/opentsdb/server.go
@@ -17,11 +17,11 @@ import (
 )
 
 var (
-	writeRequestsTCP = metrics.NewCounter(`vm_opentsdb_requests_total{name="write", net="tcp"}`)
-	writeErrorsTCP   = metrics.NewCounter(`vm_opentsdb_request_errors_total{name="write", net="tcp"}`)
+	writeRequestsTCP = metrics.NewCounter(`vm_ingestserver_requests_total{type="opentsdb", name="write", net="tcp"}`)
+	writeErrorsTCP   = metrics.NewCounter(`vm_ingestserver_request_errors_total{type="opentsdb", name="write", net="tcp"}`)
 
-	writeRequestsUDP = metrics.NewCounter(`vm_opentsdb_requests_total{name="write", net="udp"}`)
-	writeErrorsUDP   = metrics.NewCounter(`vm_opentsdb_request_errors_total{name="write", net="udp"}`)
+	writeRequestsUDP = metrics.NewCounter(`vm_ingestserver_requests_total{type="opentsdb", name="write", net="udp"}`)
+	writeErrorsUDP   = metrics.NewCounter(`vm_ingestserver_request_errors_total{type="opentsdb", name="write", net="udp"}`)
 )
 
 // Server is a server for collecting OpenTSDB TCP and UDP metrics.
diff --git a/lib/ingestserver/opentsdbhttp/server.go b/lib/ingestserver/opentsdbhttp/server.go
index 65b06af55..eeb83d9c3 100644
--- a/lib/ingestserver/opentsdbhttp/server.go
+++ b/lib/ingestserver/opentsdbhttp/server.go
@@ -14,8 +14,8 @@ import (
 )
 
 var (
-	writeRequests = metrics.NewCounter(`vm_opentsdbhttp_requests_total{name="write", net="tcp"}`)
-	writeErrors   = metrics.NewCounter(`vm_opentsdbhttp_request_errors_total{name="write", net="tcp"}`)
+	writeRequests = metrics.NewCounter(`vm_ingestserver_requests_total{type="opentsdbhttp", name="write", net="tcp"}`)
+	writeErrors   = metrics.NewCounter(`vm_ingestserver_request_errors_total{type="opentsdbhttp", name="write", net="tcp"}`)
 )
 
 // Server represents HTTP OpenTSDB server.
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index 1ea784c20..77249cb7a 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -424,7 +424,6 @@ func (stc *StaticConfig) appendScrapeWork(dst []ScrapeWork, swc *scrapeWorkConfi
 		if target == "" {
 			return nil, fmt.Errorf("`static_configs` target for `job_name` %q cannot be empty", swc.jobName)
 		}
-		target = addMissingPort(swc.scheme, target)
 		labels, err := mergeLabels(swc.jobName, swc.scheme, target, swc.metricsPath, stc.Labels, swc.externalLabels, swc.metaLabels, swc.params)
 		if err != nil {
 			return nil, fmt.Errorf("cannot merge labels for `static_configs` target for `job_name` %q: %s", swc.jobName, err)
diff --git a/lib/protoparser/influx/streamparser.go b/lib/protoparser/influx/streamparser.go
index eade73eac..c09976d17 100644
--- a/lib/protoparser/influx/streamparser.go
+++ b/lib/protoparser/influx/streamparser.go
@@ -3,7 +3,6 @@ package influx
 import (
 	"fmt"
 	"io"
-	"net/http"
 	"runtime"
 	"sync"
 	"time"
@@ -13,15 +12,14 @@ import (
 	"github.com/VictoriaMetrics/metrics"
 )
 
-// ParseStream parses req and calls callback for the parsed rows.
+// ParseStream parses r with the given args and calls callback for the parsed rows.
 //
-// The callback can be called multiple times for streamed data from req.
+// The callback can be called multiple times for streamed data from r.
 //
 // callback shouldn't hold rows after returning.
-func ParseStream(req *http.Request, callback func(db string, rows []Row) error) error {
+func ParseStream(r io.Reader, isGzipped bool, precision, db string, callback func(db string, rows []Row) error) error {
 	readCalls.Inc()
-	r := req.Body
-	if req.Header.Get("Content-Encoding") == "gzip" {
+	if isGzipped {
 		zr, err := common.GetGzipReader(r)
 		if err != nil {
 			return fmt.Errorf("cannot read gzipped influx line protocol data: %s", err)
@@ -30,9 +28,9 @@ func ParseStream(req *http.Request, callback func(db string, rows []Row) error)
 		r = zr
 	}
 
-	q := req.URL.Query()
+	// Default precision is 'ns'. See https://docs.influxdata.com/influxdb/v1.7/write_protocols/line_protocol_tutorial/#timestamp
 	tsMultiplier := int64(1e6)
-	switch q.Get("precision") {
+	switch precision {
 	case "ns":
 		tsMultiplier = 1e6
 	case "u":
@@ -47,9 +45,6 @@ func ParseStream(req *http.Request, callback func(db string, rows []Row) error)
 		tsMultiplier = -1e3 * 3600
 	}
 
-	// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
-	db := q.Get("db")
-
 	ctx := getStreamContext()
 	defer putStreamContext(ctx)
 	for ctx.Read(r, tsMultiplier) {
diff --git a/lib/protoparser/promremotewrite/streamparser.go b/lib/protoparser/promremotewrite/streamparser.go
index 986528bfb..e3576b4c5 100644
--- a/lib/protoparser/promremotewrite/streamparser.go
+++ b/lib/protoparser/promremotewrite/streamparser.go
@@ -112,7 +112,7 @@ func readSnappy(dst []byte, r io.Reader) ([]byte, error) {
 		return dst, err
 	}
 	if len(buf) > *maxInsertRequestSize {
-		return dst, fmt.Errorf("too big unpacked request; mustn't exceed `-maxInsertRequestSize=%d` bytes", *maxInsertRequestSize)
+		return dst, fmt.Errorf("too big unpacked request; mustn't exceed `-maxInsertRequestSize=%d` bytes; got %d bytes", *maxInsertRequestSize, len(buf))
 	}
 	if len(buf) > 0 && len(dst) < cap(dst) && &buf[0] == &dst[len(dst):cap(dst)][0] {
 		dst = dst[:len(dst)+len(buf)]


Test patch:
diff --git a/lib/promscrape/config_test.go b/lib/promscrape/config_test.go
index 791643064..ea139bc31 100644
--- a/lib/promscrape/config_test.go
+++ b/lib/promscrape/config_test.go
@@ -470,7 +470,7 @@ scrape_configs:
 			Labels: []prompbmarshal.Label{
 				{
 					Name:  "__address__",
-					Value: "host1:80",
+					Value: "host1",
 				},
 				{
 					Name:  "__meta_filepath",
@@ -503,7 +503,7 @@ scrape_configs:
 			Labels: []prompbmarshal.Label{
 				{
 					Name:  "__address__",
-					Value: "host2:80",
+					Value: "host2",
 				},
 				{
 					Name:  "__meta_filepath",
@@ -690,7 +690,7 @@ scrape_configs:
 			Labels: []prompbmarshal.Label{
 				{
 					Name:  "__address__",
-					Value: "foo.bar:443",
+					Value: "foo.bar",
 				},
 				{
 					Name:  "__metrics_path__",
@@ -724,7 +724,7 @@ scrape_configs:
 			Labels: []prompbmarshal.Label{
 				{
 					Name:  "__address__",
-					Value: "aaa:443",
+					Value: "aaa",
 				},
 				{
 					Name:  "__metrics_path__",
@@ -758,7 +758,7 @@ scrape_configs:
 			Labels: []prompbmarshal.Label{
 				{
 					Name:  "__address__",
-					Value: "1.2.3.4:80",
+					Value: "1.2.3.4",
 				},
 				{
 					Name:  "__metrics_path__",
@@ -1090,7 +1090,59 @@ scrape_configs:
 			TLSCertificate: &snakeoilCert,
 		},
 	})
-
+	f(`
+scrape_configs:
+  - job_name: 'snmp'
+    static_configs:
+      - targets:
+        - 192.168.1.2  # SNMP device.
+    metrics_path: /snmp
+    params:
+      module: [if_mib]
+    relabel_configs:
+      - source_labels: [__address__]
+        target_label: __param_target
+      - source_labels: [__param_target]
+        target_label: instance
+      - target_label: __address__
+        replacement: 127.0.0.1:9116  # The SNMP exporter's real hostname:port.
+`, []ScrapeWork{
+		{
+			ScrapeURL:      "http://127.0.0.1:9116/snmp?module=if_mib&target=192.168.1.2",
+			ScrapeInterval: defaultScrapeInterval,
+			ScrapeTimeout:  defaultScrapeTimeout,
+			Labels: []prompbmarshal.Label{
+				{
+					Name:  "__address__",
+					Value: "127.0.0.1:9116",
+				},
+				{
+					Name:  "__metrics_path__",
+					Value: "/snmp",
+				},
+				{
+					Name:  "__param_module",
+					Value: "if_mib",
+				},
+				{
+					Name:  "__param_target",
+					Value: "192.168.1.2",
+				},
+				{
+					Name:  "__scheme__",
+					Value: "http",
+				},
+				{
+					Name:  "instance",
+					Value: "192.168.1.2",
+				},
+				{
+					Name:  "job",
+					Value: "snmp",
+				},
+			},
+		},
+	})
 }
 
 var defaultRegexForRelabelConfig = regexp.MustCompile("^(.*)$")