app/vlselect/logsql: skip rows without _stream reference

_stream field can be empty for the recently ingested rows because respective entry in indexdb is not yet searchable as it haven't been flushed to storage yet.
This change just skips such items in the output response to make it more consistent.

See: https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6042
Signed-off-by: Zakhar Bessarab <z.bessarab@victoriametrics.com>

.github/ISSUE_TEMPLATE/bug_report.yml

@@ -5,10 +5,10 @@ body:
   - type: markdown
     attributes:
       value: |
-        Before filling a bug report it would be great to [upgrade](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-upgrade) 
+        Before filling a bug report it would be great to [upgrade](https://docs.victoriametrics.com/#how-to-upgrade) 
         to [the latest available release](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest)
         and verify whether the bug is reproducible there.
-        It's also recommended to read the [troubleshooting docs](https://docs.victoriametrics.com/victoriametrics/troubleshooting/) first.
+        It's also recommended to read the [troubleshooting docs](https://docs.victoriametrics.com/Troubleshooting.html) first.
   - type: textarea
     id: describe-the-bug
     attributes:
@@ -60,12 +60,12 @@ body:
   
         For VictoriaMetrics health-state issues please provide full-length screenshots
         of Grafana dashboards if possible:
-          * [Grafana dashboard for single-node VictoriaMetrics](https://grafana.com/grafana/dashboards/10229)
-          * [Grafana dashboard for VictoriaMetrics cluster](https://grafana.com/grafana/dashboards/11176)
+          * [Grafana dashboard for single-node VictoriaMetrics](https://grafana.com/grafana/dashboards/10229/)
+          * [Grafana dashboard for VictoriaMetrics cluster](https://grafana.com/grafana/dashboards/11176/)
         
         See how to setup monitoring here:
-          * [monitoring for single-node VictoriaMetrics](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#monitoring)
-          * [monitoring for VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#monitoring)
+          * [monitoring for single-node VictoriaMetrics](https://docs.victoriametrics.com/#monitoring)
+          * [monitoring for VictoriaMetrics cluster](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#monitoring)
     validations:
       required: false
   - type: textarea

.github/ISSUE_TEMPLATE/question.yml

@@ -5,7 +5,7 @@ body:
   - type: textarea
     id: describe-the-component
     attributes:
-      label: Is your question related to a specific component?
+      label: Is your question request related to a specific component?
       placeholder: |
         VictoriaMetrics, vmagent, vmalert, vmui, etc...
     validations:
@@ -24,9 +24,9 @@ body:
       label: Troubleshooting docs
       description: I am familiar with the following troubleshooting docs
       options:
-        - label: General - https://docs.victoriametrics.com/victoriametrics/troubleshooting/
+        - label: General - https://docs.victoriametrics.com/Troubleshooting.html
           required: false
-        - label: vmagent - https://docs.victoriametrics.com/victoriametrics/vmagent/#troubleshooting
-          required: false
-        - label: vmalert - https://docs.victoriametrics.com/victoriametrics/vmalert/#troubleshooting
+        - label: vmagent - https://docs.victoriametrics.com/vmagent.html#troubleshooting
           required: false
+        - label: vmalert - https://docs.victoriametrics.com/vmalert.html#troubleshooting
+          required: false

.github/PULL_REQUEST_TEMPLATE/pull_request_template.md

@@ -0,0 +1,35 @@
+### Describe Your Changes
+
+Please provide a brief description of the changes you made. Be as specific as possible to help others understand the purpose and impact of your modifications.
+
+### Checklist
+
+The following checks are mandatory:
+
+- [ ] I have read the [Contributing Guidelines](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/CONTRIBUTING.md)
+- [ ] All commits are signed and include `Signed-off-by` line. Use `git commit -s` to include `Signed-off-by` your commits. See this [doc](https://git-scm.com/book/en/v2/Git-Tools-Signing-Your-Work) about how to sign your commits.
+- [ ] Tests are passing locally. Use `make test` to run all tests locally.
+- [ ] Linting is passing locally. Use `make check-all` to run all linters locally.
+
+Further checks are optional for External Contributions:
+
+- [ ] Include a link to the GitHub issue in the commit message, if issue exists.
+- [ ] Mention the change in the [Changelog](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/CHANGELOG.md). Explain what has changed and why. If there is a related issue or documentation change - link them as well.
+
+  Tips for writing a good changelog message::
+
+    * Write a human-readable changelog message that describes the problem and solution.
+    * Include a link to the issue or pull request in your changelog message.
+    * Use specific language identifying the fix, such as an error message, metric name, or flag name.
+    * Provide a link to the relevant documentation for any new features you add or modify.
+
+- [ ] After your pull request is merged, please add a message to the issue with instructions for how to test the fix or try the feature you added. Here is an [example](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4048#issuecomment-1546453726)
+- [ ] Do not close the original issue before the change is released. Please note, in some cases Github can automatically close the issue once PR is merged. Re-open the issue in such case.
+- [ ] If the change somehow affects public interfaces (a new flag was added or updated, or some behavior has changed) - add the corresponding change to documentation.
+
+
+Examples of good changelog messages:
+
+1. FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add support for [VictoriaMetrics remote write protocol](https://docs.victoriametrics.com/vmagent.html#victoriametrics-remote-write-protocol) when [sending / receiving data to / from Kafka](https://docs.victoriametrics.com/vmagent.html#kafka-integration). This protocol allows saving egress network bandwidth costs when sending data from `vmagent` to `Kafka` located in another datacenter or availability zone. See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1225).
+
+2. BUGFIX: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation.html): suppress `series after dedup` error message in logs when `-remoteWrite.streamAggr.dedupInterval` command-line flag is set at [vmagent](https://docs.victoriametrics.com/vmgent.html) or when `-streamAggr.dedupInterval` command-line flag is set at [single-node VictoriaMetrics](https://docs.victoriametrics.com/).

.github/copilot-instructions.md

@@ -1,23 +0,0 @@
-# Project Overview
-
-VictoriaMetrics is a fast, cost-saving, and scalable solution for monitoring and managing time series data. It delivers high performance and reliability, making it an ideal choice for businesses of all sizes.
-
-## Folder Structure
-
-- `/app`: Contains the compilable binaries.
-- `/lib`: Contains the golang reusable libraries
-- `/docs/victoriametrics`: Contains documentation for the project.
-- `/apptest/tests`: Contains integration tests.
-
-## Libraries and Frameworks
-
-- Backend: Golang, no framework. Use third-party libraries sparingly.
-- Frontend: React.
-
-## Code review guidelines
-
-Ensure the feature or bugfix includes a changelog entry in /docs/victoriametrics/changelog/CHANGELOG.md.
-Verify the entry is under the ## tip section and matches the structure and style of existing entries.
-Chore-only changes may be omitted from the changelog.
-
-

.github/pull_request_template.md

@@ -1,10 +0,0 @@
-### Describe Your Changes
-
-Please provide a brief description of the changes you made. Be as specific as possible to help others understand the purpose and impact of your modifications.
-
-### Checklist
-
-The following checks are **mandatory**:
-
-- [ ] My change adheres to [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/victoriametrics/contributing/#pull-request-checklist).
-- [ ] My change adheres to [VictoriaMetrics development goals](https://docs.victoriametrics.com/victoriametrics/goals/).

.github/scripts/lint-changelog-tip.sh

@@ -1,48 +0,0 @@
-#!/usr/bin/env sh
-
-set -e
-
-CHANGELOG_FILE="docs/victoriametrics/changelog/CHANGELOG.md"
-
-GITHUB_BASE_REF=${GITHUB_BASE_REF:-"master"}
-GIT_REMOTE=${GIT_REMOTE:-"origin"}
-
-git diff "${GIT_REMOTE}/${GITHUB_BASE_REF}"...HEAD -- $CHANGELOG_FILE > diff.txt
-if ! grep -q "^+" diff.txt; then
-  echo "No additions in CHANGELOG.md"
-  exit 0
-fi
-
-ADDED_LINES=$(grep "^+\S" diff.txt | sed 's/^+//')
-
-START_TIP=$(grep -n "^## tip" "$CHANGELOG_FILE" | head -1 | cut -d: -f1)
-if [ -z "$START_TIP" ]; then
-  echo "ERROR: ${CHANGELOG_FILE} does not contain a ## tip section"
-  exit 1
-fi
-
-END_TIP=$(awk "NR>$START_TIP && /^## / {print NR; exit}" "${CHANGELOG_FILE}")
-if [ -z "$END_TIP" ]; then
-  END_TIP=$(wc -l < "$CHANGELOG_FILE")
-fi
-
-BAD=0
-while IFS= read -r line; do
-  # Grep exact line inside the file and get line numbers
-  MATCHES=$(grep -n -F "$line" "$CHANGELOG_FILE" | cut -d: -f1)
-  for m in $MATCHES; do
-    if [ "$m" -lt "$START_TIP" ] || [ "$m" -gt "$END_TIP" ]; then
-      echo "'$line' on line ${m} is outside ## tip section (lines ${START_TIP}-${END_TIP})"
-      BAD=1
-    fi
-  done
-done << EOF
-$ADDED_LINES
-EOF
-
-if [ "$BAD" -ne 0 ]; then
-  echo "CHANGELOG modifications must be placed inside the ## tip section."
-  exit 1
-fi
-
-echo "CHANGELOG modifications are valid."

.github/workflows/build.yml

@@ -1,80 +0,0 @@
-name: build
-
-on:
-  push:
-    branches:
-      - cluster
-      - master
-    paths:
-      - '**.go'
-      - '**/Dockerfile'
-      - '**/Makefile'
-      - '!app/vmui/**'
-      - '.github/workflows/build.yml'
-  pull_request:
-    branches:
-      - cluster
-      - master
-    paths:
-      - '**.go'
-      - '**/Dockerfile'
-      - '**/Makefile'
-      - '!app/vmui/**'
-      - '.github/workflows/build.yml'
-
-permissions:
-  contents: read
-
-concurrency:
-  cancel-in-progress: true
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-
-jobs:
-  build:
-    name: ${{ matrix.os }}-${{ matrix.arch }}
-    runs-on: ubuntu-latest
-    strategy:
-      fail-fast: false
-      matrix:
-        include:
-          - os: linux
-            arch: 386
-          - os: linux
-            arch: amd64
-          - os: linux
-            arch: arm64
-          - os: linux
-            arch: arm
-          - os: linux
-            arch: ppc64le
-          - os: linux
-            arch: s390x
-          - os: darwin
-            arch: amd64
-          - os: darwin
-            arch: arm64
-          - os: freebsd
-            arch: amd64
-          - os: openbsd
-            arch: amd64
-          - os: windows
-            arch: amd64
-    steps:
-      - name: Code checkout
-        uses: actions/checkout@v6
-
-      - name: Setup Go
-        id: go
-        uses: actions/setup-go@v6
-        with:
-          cache-dependency-path: |
-            go.sum
-            Makefile
-            app/**/Makefile
-          go-version: stable
-
-      - name: Build victoria-metrics for ${{ matrix.os }}-${{ matrix.arch }}
-        run: make victoria-metrics-${{ matrix.os }}-${{ matrix.arch }}
-
-      - name: Build vmutils for ${{ matrix.os }}-${{ matrix.arch }}
-        run: make vmutils-${{ matrix.os }}-${{ matrix.arch }}

.github/workflows/changelog-linter.yml

@@ -1,19 +0,0 @@
-name: 'changelog-linter'
-
-on:
-  pull_request:
-    paths:
-      - "docs/victoriametrics/changelog/CHANGELOG.md"
-
-jobs:
-  tip-lint:
-    runs-on: 'ubuntu-latest'
-    steps:
-      - uses: 'actions/checkout@v6'
-        with:
-          # needed for proper diff
-          fetch-depth: 0
-
-      - name: 'Validate that changelog changes are under ## tip'
-        run: |
-          GITHUB_BASE_REF=${{ github.base_ref }} ./.github/scripts/lint-changelog-tip.sh

.github/workflows/check-commit-signed.yml

@@ -1,37 +0,0 @@
-name: check-commit-signed
-
-on:
-  pull_request:
-
-jobs:
-  check-commit-signed:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@v6
-        with:
-          fetch-depth: 0  # we need full history for commit verification
-
-      - name: Check commit signatures
-        run: |
-          if [ "${{ github.event_name }}" != "pull_request" ]; then
-            echo "Not a PR event, skipping signature check"
-            exit 0
-          fi
-          
-          RANGE="${{ github.event.pull_request.base.sha }}..${{ github.event.pull_request.head.sha }}"
-          echo "Checking commits in PR range: $RANGE"
-          
-          if [ -z "$(git rev-list $RANGE)" ]; then
-            echo "No new commits in this PR, skipping signature check"
-            exit 0
-          fi
-          
-          unsigned=$(git log --pretty="%H %G?" $RANGE | grep -vE " (G|E)$" || true)
-          if [ -n "$unsigned" ]; then
-            echo "Found unsigned commits:"
-            echo "$unsigned"
-            exit 1
-          fi
-          
-          echo "All commits in PR are signed (G or E)"

.github/workflows/check-licenses.yml

@@ -19,7 +19,7 @@ jobs:
 
       - name: Setup Go
         id: go
-        uses: actions/setup-go@v6
+        uses: actions/setup-go@v5
         with:
           go-version: stable
           cache: false

.github/workflows/codeql-analysis-go.yml

@@ -1,62 +0,0 @@
-name: 'CodeQL Go'
-
-on:
-  push:
-    branches:
-      - cluster
-      - master
-    paths:
-      - '**.go'
-  pull_request:
-    branches:
-      - cluster
-      - master
-    paths:
-      - '**.go'
-
-concurrency:
-  cancel-in-progress: true
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-
-jobs:
-  analyze:
-    name: Analyze
-    runs-on: ubuntu-latest
-    permissions:
-      actions: read
-      contents: read
-      security-events: write
-
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@v6
-
-      - name: Set up Go
-        id: go
-        uses: actions/setup-go@v6
-        with:
-          cache: false
-          go-version: stable
-
-      - name: Cache Go artifacts
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.cache/go-build
-            ~/go/bin
-            ~/go/pkg/mod
-          key: go-artifacts-${{ runner.os }}-codeql-analyze-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
-          restore-keys: go-artifacts-${{ runner.os }}-codeql-analyze-
-
-      - name: Initialize CodeQL
-        uses: github/codeql-action/init@v4
-        with:
-          languages: go
-
-      - name: Autobuild
-        uses: github/codeql-action/autobuild@v4
-
-      - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@v4
-        with:
-          category: 'language:go'

.github/workflows/codeql-analysis-js.yml

@@ -0,0 +1,46 @@
+name: "CodeQL - JS"
+
+on:
+  push:
+    branches: [master, cluster]
+    paths:
+      - "**.js"
+  pull_request:
+    # The branches below must be a subset of the branches above
+    branches: [master, cluster]
+    paths:
+      - "**.js"
+  schedule:
+    - cron: "30 18 * * 2"
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+  
+jobs:
+  analyze:
+    name: Analyze
+    runs-on: ubuntu-latest
+    permissions:
+      actions: read
+      contents: read
+      security-events: write
+
+    strategy:
+      fail-fast: false
+      matrix:
+        language: ["javascript"]
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Initialize CodeQL
+        uses: github/codeql-action/init@v3
+        with:
+          languages: ${{ matrix.language }}
+
+      - name: Perform CodeQL Analysis
+        uses: github/codeql-action/analyze@v3
+        with:
+          category: "javascript"

.github/workflows/codeql-analysis.yml

@@ -0,0 +1,103 @@
+# For most projects, this workflow file will not need changing; you simply need
+# to commit it to your repository.
+#
+# You may wish to alter this file to override the set of languages analyzed,
+# or to provide custom queries or build logic.
+#
+# ******** NOTE ********
+# We have attempted to detect the languages in your repository. Please check
+# the `language` matrix defined below to confirm you have the correct set of
+# supported CodeQL languages.
+#
+name: "CodeQL"
+
+on:
+  push:
+    branches: [master, cluster]
+    paths-ignore:
+      - "docs/**"
+      - "**.md"
+      - "**.txt"
+      - "**.js"
+  pull_request:
+    # The branches below must be a subset of the branches above
+    branches: [master, cluster]
+    paths-ignore:
+      - "docs/**"
+      - "**.md"
+      - "**.txt"
+      - "**.js"
+  schedule:
+    - cron: "30 18 * * 2"
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  analyze:
+    name: Analyze
+    runs-on: ubuntu-latest
+    permissions:
+      actions: read
+      contents: read
+      security-events: write
+
+    strategy:
+      fail-fast: false
+      matrix:
+        language: ["go"]
+        # CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python', 'ruby' ]
+        # Learn more about CodeQL language support at https://git.io/codeql-language-support
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Go
+        id: go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+          cache: false
+        if: ${{ matrix.language == 'go' }}
+
+      - name: Cache Go artifacts
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cache/go-build
+            ~/go/pkg/mod
+            ~/go/bin
+          key: go-artifacts-${{ runner.os }}-codeql-analyze-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
+          restore-keys: go-artifacts-${{ runner.os }}-codeql-analyze-
+        if: ${{ matrix.language == 'go' }}
+
+      # Initializes the CodeQL tools for scanning.
+      - name: Initialize CodeQL
+        uses: github/codeql-action/init@v3
+        with:
+          languages: ${{ matrix.language }}
+          # If you wish to specify custom queries, you can do so here or in a config file.
+          # By default, queries listed here will override any specified in a config file.
+          # Prefix the list here with "+" to use these queries and those in the config file.
+          # queries: ./path/to/local/query, your-org/your-repo/queries@main
+
+      # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
+      # If this step fails, then you should remove it and run the build manually (see below)
+      - name: Autobuild
+        uses: github/codeql-action/autobuild@v3
+
+      # ℹ️ Command-line programs to run using the OS shell.
+      # 📚 https://git.io/JvXDl
+
+      # ✏️ If the Autobuild fails above, remove it and uncomment the following three lines
+      #    and modify them (or add more) to build your code if your project
+      #    uses a compiled language
+
+      #- run: |
+      #   make bootstrap
+      #   make release
+
+      - name: Perform CodeQL Analysis
+        uses: github/codeql-action/analyze@v3

.github/workflows/docs.yaml

@@ -1,57 +0,0 @@
-name: publish-docs
-on:
-  push:
-    branches:
-      - 'master'
-    paths:
-      - 'docs/**'
-      - '.github/workflows/docs.yaml'
-  workflow_dispatch: {}
-permissions:
-  contents: read  # This is required for actions/checkout and to commit back image update
-  deployments: write
-jobs:
-  build:
-    name: Build
-    runs-on: ubuntu-latest
-    steps:
-      - name: Code checkout
-        uses: actions/checkout@v6
-        with:
-          path: __vm
-
-      - name: Checkout private code
-        uses: actions/checkout@v6
-        with:
-          repository: VictoriaMetrics/vmdocs
-          token: ${{ secrets.VM_BOT_GH_TOKEN }}
-          path: __vm-docs
-
-      - name: Import GPG key
-        uses: crazy-max/ghaction-import-gpg@v6
-        id: import-gpg
-        with:
-          gpg_private_key: ${{ secrets.VM_BOT_GPG_PRIVATE_KEY }}
-          passphrase: ${{ secrets.VM_BOT_PASSPHRASE }}
-          git_user_signingkey: true
-          git_commit_gpgsign: true
-          git_config_global: true
-
-      - name: Copy docs
-        id: update
-        run: |
-          find docs -type d -maxdepth 1 -mindepth 1 -exec \
-            sh -c 'rsync -zarvh --delete {}/ ../__vm-docs/content/$(basename {})/' \;
-          echo "SHORT_SHA=$(git rev-parse --short $GITHUB_SHA)" >> $GITHUB_OUTPUT
-        working-directory: __vm
-
-      - name: Push to vmdocs
-        run: |
-          git config --global user.name "${{ steps.import-gpg.outputs.email }}"
-          git config --global user.email "${{ steps.import-gpg.outputs.email }}"
-          if [[ -n $(git status --porcelain) ]]; then
-            git add .
-            git commit -S -m "sync docs with VictoriaMetrics/VictoriaMetrics commit: ${{ steps.update.outputs.SHORT_SHA }}"
-            git push
-          fi
-        working-directory: __vm-docs

.github/workflows/main.yml

@@ -0,0 +1,120 @@
+name: main
+on:
+  push:
+    branches:
+      - master
+      - cluster
+    paths-ignore:
+      - "docs/**"
+      - "**.md"
+      - "dashboards/**"
+      - "deployment/**.yml"
+  pull_request:
+    branches:
+      - master
+      - cluster
+    paths-ignore:
+      - "docs/**"
+      - "**.md"
+      - "dashboards/**"
+      - "deployment/**.yml"
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  lint:
+    name: lint
+    runs-on: ubuntu-latest
+    steps:
+      - name: Code checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Go
+        id: go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+          cache: false
+
+      - name: Cache Go artifacts
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cache/go-build
+            ~/go/pkg/mod
+            ~/go/bin
+          key: go-artifacts-${{ runner.os }}-check-all-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
+          restore-keys: go-artifacts-${{ runner.os }}-check-all-
+
+      - name: Run check-all
+        run: |
+          make check-all
+          git diff --exit-code
+
+  build:
+    needs: lint
+    name: build
+    runs-on: ubuntu-latest
+    steps:
+      - name: Code checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Go
+        id: go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+          cache: false
+
+      - name: Cache Go artifacts
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cache/go-build
+            ~/go/pkg/mod
+            ~/go/bin
+          key: go-artifacts-${{ runner.os }}-crossbuild-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
+          restore-keys: go-artifacts-${{ runner.os }}-crossbuild-
+
+      - name: Build
+        run: make crossbuild
+
+  test:
+    needs: lint
+    strategy:
+      matrix:
+        scenario: ["test-full", "test-pure", "test-full-386"]
+    name: test
+    runs-on: ubuntu-latest
+    steps:
+      - name: Code checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Go
+        id: go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+          cache: false
+
+      - name: Cache Go artifacts
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cache/go-build
+            ~/go/pkg/mod
+            ~/go/bin
+          key: go-artifacts-${{ runner.os }}-${{ matrix.scenario }}-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
+          restore-keys: go-artifacts-${{ runner.os }}-${{ matrix.scenario }}-
+
+      - name: run tests
+        run: make ${{ matrix.scenario}}
+
+      - name: Publish coverage
+        uses: codecov/codecov-action@v4
+        with:
+          file: ./coverage.txt

.github/workflows/sync-docs.yml

@@ -0,0 +1,51 @@
+name: publish-docs
+on:
+  push:
+    branches:
+      - 'master'
+    paths:
+      - 'docs/**'
+  workflow_dispatch: {}
+permissions:
+  contents: read  # This is required for actions/checkout and to commit back image update
+  deployments: write
+jobs:
+  build:
+    name: Build
+    runs-on: ubuntu-latest
+    steps:
+      - name: Code checkout
+        uses: actions/checkout@v4
+        with:
+          path: main
+      - name: Checkout private code
+        uses: actions/checkout@v4
+        with:
+          repository: VictoriaMetrics/vmdocs
+          token: ${{ secrets.VM_BOT_GH_TOKEN }}
+          path: docs
+      - name: Import GPG key
+        uses: crazy-max/ghaction-import-gpg@v5
+        with:
+          gpg_private_key: ${{ secrets.VM_BOT_GPG_PRIVATE_KEY }}
+          passphrase: ${{ secrets.VM_BOT_PASSPHRASE }}
+          git_user_signingkey: true
+          git_commit_gpgsign: true
+          workdir: docs
+      - name: Set short git commit SHA
+        id: vars
+        run: |
+          calculatedSha=$(git rev-parse --short ${{ github.sha }})
+          echo "short_sha=$calculatedSha" >> $GITHUB_OUTPUT
+        working-directory: main
+      - name: update code and commit
+        run: |
+          rm -rf content
+          cp -r ../main/docs content
+          make clean-after-copy
+          git config --global user.name "${{ steps.import-gpg.outputs.email }}"
+          git config --global user.email "${{ steps.import-gpg.outputs.email }}"
+          git add .
+          git commit -S -m "sync docs with VictoriaMetrics/VictoriaMetrics commit: ${{ steps.vars.outputs.short_sha }}"
+          git push
+        working-directory: docs

.github/workflows/test.yml

@@ -1,113 +0,0 @@
-name: test
-
-on:
-  push:
-    branches:
-      - cluster
-      - master
-    paths:
-      - '**.go'
-      - 'go.*'
-      - '.github/workflows/main.yml'
-  pull_request:
-    branches:
-      - cluster
-      - master
-    paths:
-      - '**.go'
-      - 'go.*'
-      - '.github/workflows/main.yml'
-
-permissions:
-  contents: read
-
-concurrency:
-  cancel-in-progress: true
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-
-
-jobs:
-  lint:
-    name: lint
-    runs-on: ubuntu-latest
-    steps:
-      - name: Code checkout
-        uses: actions/checkout@v6
-
-      - name: Setup Go
-        id: go
-        uses: actions/setup-go@v6
-        with:
-          cache-dependency-path: |
-            go.sum
-            Makefile
-            app/**/Makefile
-          go-version: stable
-
-
-      - name: Cache golangci-lint
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.cache/golangci-lint
-            ~/go/bin
-          key: golangci-lint-${{ runner.os }}-${{ hashFiles('.golangci.yml') }}
-
-      - name: Run check-all
-        run: |
-          make check-all
-          git diff --exit-code
-
-  unit:
-    name: unit
-    runs-on: ubuntu-latest
-
-    strategy:
-      matrix:
-        scenario:
-          - 'test-full'
-          - 'test-full-386'
-          - 'test-pure'
-
-    steps:
-      - name: Code checkout
-        uses: actions/checkout@v6
-
-      - name: Setup Go
-        id: go
-        uses: actions/setup-go@v6
-        with:
-          cache-dependency-path: |
-            go.sum
-            Makefile
-            app/**/Makefile
-          go-version: stable
-
-      - name: Run tests
-        run: GOGC=10 make ${{ matrix.scenario}}
-
-      - name: Publish coverage
-        uses: codecov/codecov-action@v5
-        with:
-          files: ./coverage.txt
-
-  integration:
-    name: integration
-    runs-on: ubuntu-latest
-
-    steps:
-      - name: Code checkout
-        uses: actions/checkout@v6
-
-      - name: Setup Go
-        id: go
-        uses: actions/setup-go@v6
-        with:
-          cache-dependency-path: |
-            go.sum
-            Makefile
-            app/**/Makefile
-          go-version: stable
-
-      - name: Run integration tests
-        run: make integration-test

.github/workflows/vmui.yml

@@ -1,88 +0,0 @@
-name: vmui
-
-on:
-  push:
-    branches:
-      - cluster
-      - master
-    paths:
-      - 'app/vmui/packages/vmui/**'
-      - '.github/workflows/vmui.yml'
-  pull_request:
-    branches:
-      - cluster
-      - master
-    paths:
-      - 'app/vmui/packages/vmui/**'
-      - '.github/workflows/vmui.yml'
-
-permissions:
-  contents: read
-  packages: read
-  pull-requests: read
-  checks: write
-
-concurrency:
-  cancel-in-progress: true
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-
-jobs:
-  vmui-checks:
-    name: VMUI Checks (lint, test, typecheck)
-    runs-on: ubuntu-latest
-    steps:
-      - name: Code checkout
-        uses: actions/checkout@v6
-
-      - name: Cache node_modules
-        id: cache
-        uses: actions/cache@v5
-        with:
-          path: app/vmui/packages/vmui/node_modules
-          key: vmui-deps-${{ runner.os }}-${{ hashFiles('app/vmui/packages/vmui/package-lock.json', 'app/vmui/Dockerfile-build') }}
-          restore-keys: |
-            vmui-deps-${{ runner.os }}-
-
-      - name: Install dependencies
-        if: steps.cache.outputs.cache-hit != 'true'
-        run: make vmui-install
-
-      - name: Run lint
-        id: lint
-        run: make vmui-lint
-        continue-on-error: true
-        env:
-          VMUI_SKIP_INSTALL: true
-
-      - name: Run tests
-        id: test
-        run: make vmui-test
-        continue-on-error: true
-        env:
-          VMUI_SKIP_INSTALL: true
-
-      - name: Run typecheck
-        id: typecheck
-        run: make vmui-typecheck
-        continue-on-error: true
-        env:
-          VMUI_SKIP_INSTALL: true
-
-      - name: Annotate Code Linting Results
-        uses: ataylorme/eslint-annotate-action@v3
-        with:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          report-json: app/vmui/packages/vmui/vmui-lint-report.json
-
-      - name: Check overall status
-        run: |
-          echo "Lint status: ${{ steps.lint.outcome }}"
-          echo "Test status: ${{ steps.test.outcome }}"
-          echo "Typecheck status: ${{ steps.typecheck.outcome }}"
-
-          if [[ "${{ steps.lint.outcome }}" == "failure" || "${{ steps.test.outcome }}" == "failure" || "${{ steps.typecheck.outcome }}" == "failure" ]]; then
-            echo "One or more checks failed"
-            exit 1
-          else
-            echo "All checks passed"
-          fi

.gitignore

@@ -7,12 +7,10 @@
 .vscode
 *.test
 *.swp
-/vmdocs
 /gocache-for-docker
 /victoria-logs-data
 /victoria-metrics-data
 /vmagent-remotewrite-data
-/vlagent-remotewritewrite
 /vmstorage-data
 /vmselect-cache
 /package/temp-deb-*
@@ -24,8 +22,4 @@ Gemfile.lock
 /_site
 _site
 *.tmp
-/docs/.jekyll-metadata
-coverage.txt
-cspell.json
-*~
-deployment/docker/provisioning/plugins/
+/docs/.jekyll-metadata

.golangci.yml

@@ -1,29 +1,19 @@
-version: "2"
+run:
+  timeout: 2m
+
 linters:
-  settings:
-    errcheck:
-      exclude-functions:
-        - fmt.Fprintf
-        - fmt.Fprint
-        - (net/http.ResponseWriter).Write
-  exclusions:
-    generated: lax
-    presets:
-      - common-false-positives
-      - legacy
-      - std-error-handling
-    rules:
-      - linters:
-          - staticcheck
-        text: 'SA(4003|1019|5011):'
-    paths:
-      - third_party$
-      - builtin$
-      - examples$
-formatters:
-  exclusions:
-    generated: lax
-    paths:
-      - third_party$
-      - builtin$
-      - examples$
+  enable:
+  - revive
+
+issues:
+  exclude-rules:
+  - linters:
+    - staticcheck
+    text: "SA(4003|1019|5011):"
+  include:
+  - EXC0012
+  - EXC0014
+
+linters-settings:
+  errcheck:
+    exclude: ./errcheck_excludes.txt

.wwhrd.yml

@@ -4,4 +4,3 @@ allowlist:
   - BSD-3-Clause
   - BSD-2-Clause
   - ISC
-  - MPL-2.0

CODE_OF_CONDUCT_RU.md

@@ -0,0 +1,120 @@
+
+# Кодекс Поведения участника
+
+## Наши обязательства
+
+Мы, как участники, авторы и лидеры обязуемся сделать участие в сообществе
+свободным от притеснений для всех, независимо от возраста, телосложения,
+видимых или невидимых ограничений способности, этнической принадлежности,
+половых признаков, гендерной идентичности и выражения, уровня опыта,
+образования, социо-экономического статуса, национальности, внешности,
+расы, религии, или сексуальной идентичности и ориентации.
+
+Мы обещаем действовать и взаимодействовать таким образом, чтобы вносить вклад в открытое,
+дружелюбное, многообразное, инклюзивное и здоровое сообщество.
+
+## Наши стандарты
+
+Примеры поведения, создающие условия для благоприятных взаимоотношений включают в себя:
+
+* Проявление доброты и эмпатии к другим участникам проекта
+* Уважение к чужой точке зрения и опыту
+* Конструктивная критика и принятие конструктивной критики
+* Принятие ответственности, принесение извинений тем, кто пострадал от наших ошибок
+  и извлечение уроков из опыта
+* Ориентирование на то, что лучше подходит для сообщества, а не только для нас лично
+
+Примеры неприемлемого поведения участников включают в себя:
+
+* Использование выражений или изображений сексуального характера и нежелательное сексуальное внимание или домогательство в любой форме
+* Троллинг, оскорбительные или уничижительные комментарии, переход на личности или затрагивание политических убеждений
+* Публичное или приватное домогательство
+* Публикация личной информации других лиц, например, физического или электронного адреса, без явного разрешения
+* Иное поведение, которое обоснованно считать неуместным в профессиональной обстановке
+
+## Обязанности
+
+Лидеры сообщества отвечают за разъяснение и применение наших стандартов приемлемого
+поведения и будут предпринимать соответствующие и честные меры по исправлению положения
+в ответ на любое поведение, которое они сочтут неприемлемым, угрожающим, оскорбительным или вредным.
+
+Лидеры сообщества обладают правом и обязанностью удалять, редактировать или отклонять
+комментарии, коммиты, код, изменения в вики, вопросы и другой вклад, который не совпадает
+с Кодексом Поведения, и предоставят причины принятого решения, когда сочтут нужным.
+
+## Область применения
+
+Данный Кодекс Поведения применим во всех во всех публичных физических и цифровых пространства сообщества,
+а также когда человек официально представляет сообщество в публичных местах.
+Примеры представления проекта или сообщества включают использование официальной электронной почты,
+публикации в официальном аккаунте в социальных сетях,
+или упоминания как представителя в онлайн или оффлайн мероприятии.
+
+## Приведение в исполнение
+
+О случаях домогательства, а так же оскорбительного или иного другого неприемлемого
+поведения можно сообщить ответственным лидерам сообщества с помощью письма на info@victoriametrics.com
+Все жалобы будут рассмотрены и расследованы оперативно и беспристрастно.
+
+Все лидеры сообщества обязаны уважать неприкосновенность частной жизни и личную
+неприкосновенность автора сообщения.
+
+## Руководство по исполнению
+
+Лидеры сообщества будут следовать следующим Принципам Воздействия в Сообществе,
+чтобы определить последствия для тех, кого они считают виновными в нарушении данного Кодекса Поведения:
+
+### 1. Исправление
+
+**Общественное влияние**: Использование недопустимой лексики или другое поведение,
+считающиеся непрофессиональным или нежелательным в сообществе.
+
+**Последствия**: Личное, письменное предупреждение от лидеров сообщества,
+объясняющее суть нарушения и почему такое поведение
+было неуместно. Лидеры сообщества могут попросить принести публичное извинение.
+
+### 2. Предупреждение
+
+**Общественное влияние**: Нарушение в результате одного инцидента или серии действий.
+
+**Последствия**: Предупреждение о последствиях в случае продолжающегося неуместного поведения.
+На определенное время не допускается взаимодействие с людьми, вовлеченными в инцидент,
+включая незапрошенное взаимодействие
+с теми, кто обеспечивает соблюдение Кодекса. Это включает в себя избегание взаимодействия
+в публичных пространствах, а так же во внешних каналах,
+таких как социальные сети. Нарушение этих правил влечет за собой временный или вечный бан.
+
+### 3. Временный бан
+
+**Общественное влияние**: Серьёзное нарушение стандартов сообщества,
+включая продолжительное неуместное поведение.
+
+**Последствия**: Временный запрет (бан) на любое взаимодействие
+или публичное общение с сообществом на определенный период времени.
+На этот период не допускается публичное или личное взаимодействие с людьми,
+вовлеченными в инцидент, включая незапрошенное взаимодействие
+с теми, кто обеспечивает соблюдение Кодекса.
+Нарушение этих правил влечет за собой вечный бан.
+
+### 4. Вечный бан
+
+**Общественное влияние**: Демонстрация систематических нарушений стандартов сообщества,
+включая продолжающееся неуместное поведение, домогательство до отдельных лиц,
+или проявление агрессии либо пренебрежительного отношения к категориям лиц.
+
+**Последствия**: Вечный запрет на любое публичное взаимодействие с сообществом.
+
+## Атрибуция
+
+Данный Кодекс Поведения основан на [Кодекс Поведения участника][homepage],
+версии 2.0, доступной по адресу
+<https://www.contributor-covenant.org/version/2/0/code_of_conduct.html>.
+
+Принципы Воздействия в Сообществе были вдохновлены [Mozilla's code of conduct
+enforcement ladder](https://github.com/mozilla/diversity).
+
+[homepage]: https://www.contributor-covenant.org
+
+Ответы на общие вопросы о данном кодексе поведения ищите на странице FAQ:
+<https://www.contributor-covenant.org/faq>. Переводы доступны по адресу
+<https://www.contributor-covenant.org/translations>.

CONTRIBUTING.md

@@ -1 +1,21 @@
-The document has been moved [here](https://docs.victoriametrics.com/victoriametrics/contributing/).
+If you like VictoriaMetrics and want to contribute, then we need the following:
+
+- Filing issues and feature requests [here](https://github.com/VictoriaMetrics/VictoriaMetrics/issues).
+- Spreading a word about VictoriaMetrics: conference talks, articles, comments, experience sharing with colleagues.
+- Updating documentation.
+
+We are open to third-party pull requests provided they follow [KISS design principle](https://en.wikipedia.org/wiki/KISS_principle):
+
+- Prefer simple code and architecture.
+- Avoid complex abstractions.
+- Avoid magic code and fancy algorithms.
+- Avoid [big external dependencies](https://medium.com/@valyala/stripping-dependency-bloat-in-victoriametrics-docker-image-983fb5912b0d).
+- Minimize the number of moving parts in the distributed system.
+- Avoid automated decisions, which may hurt cluster availability, consistency or performance.
+
+Adhering `KISS` principle simplifies the resulting code and architecture, so it can be reviewed, understood and verified by many people.
+
+Before sending a pull request please check the following:
+- [ ] All commits are signed and include `Signed-off-by` line. Use `git commit -s` to include `Signed-off-by` your commits. See this [doc](https://git-scm.com/book/en/v2/Git-Tools-Signing-Your-Work) about how to sign your commits.
+- [ ] Tests are passing locally. Use `make test` to run all tests locally.
+- [ ] Linting is passing locally. Use `make check-all` to run all linters locally.

LICENSE

@@ -175,7 +175,7 @@
 
    END OF TERMS AND CONDITIONS
 
-   Copyright 2019-2026 VictoriaMetrics, Inc.
+   Copyright 2019-2024 VictoriaMetrics, Inc.
 
    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.

Makefile

@@ -5,32 +5,27 @@ MAKE_PARALLEL := $(MAKE) -j $(MAKE_CONCURRENCY)
 DATEINFO_TAG ?= $(shell date -u +'%Y%m%d-%H%M%S')
 BUILDINFO_TAG ?= $(shell echo $$(git describe --long --all | tr '/' '-')$$( \
 	      git diff-index --quiet HEAD -- || echo '-dirty-'$$(git diff-index -u HEAD | openssl sha1 | cut -d' ' -f2 | cut -c 1-8)))
+LATEST_TAG ?= latest
 
 PKG_TAG ?= $(shell git tag -l --points-at HEAD)
 ifeq ($(PKG_TAG),)
 PKG_TAG := $(BUILDINFO_TAG)
 endif
 
-EXTRA_DOCKER_TAG_SUFFIX ?=
-EXTRA_GO_BUILD_TAGS ?=
-
 GO_BUILDINFO = -X '$(PKG_PREFIX)/lib/buildinfo.Version=$(APP_NAME)-$(DATEINFO_TAG)-$(BUILDINFO_TAG)'
-TAR_OWNERSHIP ?= --owner=1000 --group=1000
-
-GOLANGCI_LINT_VERSION := 2.7.2
 
 .PHONY: $(MAKECMDGOALS)
 
 include app/*/Makefile
-include codespell/Makefile
 include docs/Makefile
 include deployment/*/Makefile
 include dashboards/Makefile
+include snap/local/Makefile
 include package/release/Makefile
-include benchmarks/Makefile
 
 all: \
 	victoria-metrics-prod \
+	victoria-logs-prod \
 	vmagent-prod \
 	vmalert-prod \
 	vmalert-tool-prod \
@@ -54,6 +49,7 @@ publish: \
 
 package: \
 	package-victoria-metrics \
+	package-victoria-logs \
 	package-vmagent \
 	package-vmalert \
 	package-vmalert-tool \
@@ -125,15 +121,6 @@ vmutils-linux-ppc64le: \
 	vmrestore-linux-ppc64le \
 	vmctl-linux-ppc64le
 
-vmutils-linux-s390x: \
-	vmagent-linux-s390x \
-	vmalert-linux-s390x \
-	vmalert-tool-linux-s390x \
-	vmauth-linux-s390x \
-	vmbackup-linux-s390x \
-	vmrestore-linux-s390x \
-	vmctl-linux-s390x
-
 vmutils-darwin-amd64: \
 	vmagent-darwin-amd64 \
 	vmalert-darwin-amd64 \
@@ -179,11 +166,9 @@ vmutils-windows-amd64: \
 	vmrestore-windows-amd64 \
 	vmctl-windows-amd64
 
-# When adding a new crossbuild target, please also add it to the .github/workflows/build.yml
 crossbuild:
 	$(MAKE_PARALLEL) victoria-metrics-crossbuild vmutils-crossbuild
 
-# When adding a new crossbuild target, please also add it to the .github/workflows/build.yml
 victoria-metrics-crossbuild: \
 	victoria-metrics-linux-386 \
 	victoria-metrics-linux-amd64 \
@@ -196,7 +181,6 @@ victoria-metrics-crossbuild: \
 	victoria-metrics-openbsd-amd64 \
 	victoria-metrics-windows-amd64
 
-# When adding a new crossbuild target, please also add it to the .github/workflows/build.yml
 vmutils-crossbuild: \
 	vmutils-linux-386 \
 	vmutils-linux-amd64 \
@@ -209,52 +193,12 @@ vmutils-crossbuild: \
 	vmutils-openbsd-amd64 \
 	vmutils-windows-amd64
 
-publish-final-images:
-	PKG_TAG=$(TAG) APP_NAME=victoria-metrics $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG) APP_NAME=vmagent $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG) APP_NAME=vmalert $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG) APP_NAME=vmalert-tool $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG) APP_NAME=vmauth $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG) APP_NAME=vmbackup $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG) APP_NAME=vmrestore $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG) APP_NAME=vmctl $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-cluster APP_NAME=vminsert $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-cluster APP_NAME=vmselect $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-cluster APP_NAME=vmstorage $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise APP_NAME=victoria-metrics $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise APP_NAME=vmagent $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise APP_NAME=vmalert $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise APP_NAME=vmauth $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise APP_NAME=vmbackup $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise APP_NAME=vmrestore $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise-cluster APP_NAME=vminsert $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise-cluster APP_NAME=vmselect $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise-cluster APP_NAME=vmstorage $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise APP_NAME=vmgateway $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG)-enterprise APP_NAME=vmbackupmanager $(MAKE) publish-via-docker-from-rc && \
-	PKG_TAG=$(TAG) $(MAKE) publish-latest
-
-publish-latest:
-	PKG_TAG=$(TAG) APP_NAME=victoria-metrics $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG) APP_NAME=vmagent $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG) APP_NAME=vmalert $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG) APP_NAME=vmalert-tool $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG) APP_NAME=vmauth $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG) APP_NAME=vmbackup $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG) APP_NAME=vmrestore $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG) APP_NAME=vmctl $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG)-cluster APP_NAME=vminsert $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG)-cluster APP_NAME=vmselect $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG)-cluster APP_NAME=vmstorage $(MAKE) publish-via-docker-latest && \
-	PKG_TAG=$(TAG)-enterprise APP_NAME=vmgateway $(MAKE) publish-via-docker-latest
-	PKG_TAG=$(TAG)-enterprise APP_NAME=vmbackupmanager $(MAKE) publish-via-docker-latest
-
 publish-release:
 	rm -rf bin/*
-	git checkout $(TAG) && $(MAKE) release && $(MAKE) publish && \
-		git checkout $(TAG)-cluster && $(MAKE) release && $(MAKE) publish && \
-		git checkout $(TAG)-enterprise && $(MAKE) release && $(MAKE) publish && \
-		git checkout $(TAG)-enterprise-cluster && $(MAKE) release && $(MAKE) publish
+	git checkout $(TAG) && $(MAKE) release && LATEST_TAG=stable $(MAKE) publish && \
+		git checkout $(TAG)-cluster && $(MAKE) release && LATEST_TAG=cluster-stable $(MAKE) publish && \
+		git checkout $(TAG)-enterprise && $(MAKE) release && LATEST_TAG=enterprise-stable $(MAKE) publish && \
+		git checkout $(TAG)-enterprise-cluster && $(MAKE) release && LATEST_TAG=enterprise-cluster-stable $(MAKE) publish
 
 release:
 	$(MAKE_PARALLEL) \
@@ -266,7 +210,6 @@ release-victoria-metrics: \
 	release-victoria-metrics-linux-amd64 \
 	release-victoria-metrics-linux-arm \
 	release-victoria-metrics-linux-arm64 \
-	release-victoria-metrics-linux-s390x \
 	release-victoria-metrics-darwin-amd64 \
 	release-victoria-metrics-darwin-arm64 \
 	release-victoria-metrics-freebsd-amd64 \
@@ -285,9 +228,6 @@ release-victoria-metrics-linux-arm:
 release-victoria-metrics-linux-arm64:
 	GOOS=linux GOARCH=arm64 $(MAKE) release-victoria-metrics-goos-goarch
 
-release-victoria-metrics-linux-s390x:
-	GOOS=linux GOARCH=s390x $(MAKE) release-victoria-metrics-goos-goarch
-
 release-victoria-metrics-darwin-amd64:
 	GOOS=darwin GOARCH=amd64 $(MAKE) release-victoria-metrics-goos-goarch
 
@@ -305,7 +245,7 @@ release-victoria-metrics-windows-amd64:
 
 release-victoria-metrics-goos-goarch: victoria-metrics-$(GOOS)-$(GOARCH)-prod
 	cd bin && \
-		tar $(TAR_OWNERSHIP) --transform="flags=r;s|-$(GOOS)-$(GOARCH)||" -czf victoria-metrics-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
+		tar --transform="flags=r;s|-$(GOOS)-$(GOARCH)||" -czf victoria-metrics-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
 			victoria-metrics-$(GOOS)-$(GOARCH)-prod \
 		&& sha256sum victoria-metrics-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
 			victoria-metrics-$(GOOS)-$(GOARCH)-prod \
@@ -322,12 +262,68 @@ release-victoria-metrics-windows-goarch: victoria-metrics-windows-$(GOARCH)-prod
 	cd bin && rm -rf \
 		victoria-metrics-windows-$(GOARCH)-prod.exe
 
+release-victoria-logs:
+	$(MAKE_PARALLEL) release-victoria-logs-linux-386 \
+		release-victoria-logs-linux-amd64 \
+		release-victoria-logs-linux-arm \
+		release-victoria-logs-linux-arm64 \
+		release-victoria-logs-darwin-amd64 \
+		release-victoria-logs-darwin-arm64 \
+		release-victoria-logs-freebsd-amd64 \
+		release-victoria-logs-openbsd-amd64 \
+		release-victoria-logs-windows-amd64
+
+release-victoria-logs-linux-386:
+	GOOS=linux GOARCH=386 $(MAKE) release-victoria-logs-goos-goarch
+
+release-victoria-logs-linux-amd64:
+	GOOS=linux GOARCH=amd64 $(MAKE) release-victoria-logs-goos-goarch
+
+release-victoria-logs-linux-arm:
+	GOOS=linux GOARCH=arm $(MAKE) release-victoria-logs-goos-goarch
+
+release-victoria-logs-linux-arm64:
+	GOOS=linux GOARCH=arm64 $(MAKE) release-victoria-logs-goos-goarch
+
+release-victoria-logs-darwin-amd64:
+	GOOS=darwin GOARCH=amd64 $(MAKE) release-victoria-logs-goos-goarch
+
+release-victoria-logs-darwin-arm64:
+	GOOS=darwin GOARCH=arm64 $(MAKE) release-victoria-logs-goos-goarch
+
+release-victoria-logs-freebsd-amd64:
+	GOOS=freebsd GOARCH=amd64 $(MAKE) release-victoria-logs-goos-goarch
+
+release-victoria-logs-openbsd-amd64:
+	GOOS=openbsd GOARCH=amd64 $(MAKE) release-victoria-logs-goos-goarch
+
+release-victoria-logs-windows-amd64:
+	GOARCH=amd64 $(MAKE) release-victoria-logs-windows-goarch
+
+release-victoria-logs-goos-goarch: victoria-logs-$(GOOS)-$(GOARCH)-prod
+	cd bin && \
+		tar --transform="flags=r;s|-$(GOOS)-$(GOARCH)||" -czf victoria-logs-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
+			victoria-logs-$(GOOS)-$(GOARCH)-prod \
+		&& sha256sum victoria-logs-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
+			victoria-logs-$(GOOS)-$(GOARCH)-prod \
+			| sed s/-$(GOOS)-$(GOARCH)-prod/-prod/ > victoria-logs-$(GOOS)-$(GOARCH)-$(PKG_TAG)_checksums.txt
+	cd bin && rm -rf victoria-logs-$(GOOS)-$(GOARCH)-prod
+
+release-victoria-logs-windows-goarch: victoria-logs-windows-$(GOARCH)-prod
+	cd bin && \
+		zip victoria-logs-windows-$(GOARCH)-$(PKG_TAG).zip \
+			victoria-logs-windows-$(GOARCH)-prod.exe \
+		&& sha256sum victoria-logs-windows-$(GOARCH)-$(PKG_TAG).zip \
+			victoria-logs-windows-$(GOARCH)-prod.exe \
+			> victoria-logs-windows-$(GOARCH)-$(PKG_TAG)_checksums.txt
+	cd bin && rm -rf \
+		victoria-logs-windows-$(GOARCH)-prod.exe
+
 release-vmutils: \
 	release-vmutils-linux-386 \
 	release-vmutils-linux-amd64 \
 	release-vmutils-linux-arm64 \
 	release-vmutils-linux-arm \
-	release-vmutils-linux-s390x \
 	release-vmutils-darwin-amd64 \
 	release-vmutils-darwin-arm64 \
 	release-vmutils-freebsd-amd64 \
@@ -336,7 +332,7 @@ release-vmutils: \
 
 release-vmutils-linux-386:
 	GOOS=linux GOARCH=386 $(MAKE) release-vmutils-goos-goarch
-
+	
 release-vmutils-linux-amd64:
 	GOOS=linux GOARCH=amd64 $(MAKE) release-vmutils-goos-goarch
 
@@ -346,9 +342,6 @@ release-vmutils-linux-arm64:
 release-vmutils-linux-arm:
 	GOOS=linux GOARCH=arm $(MAKE) release-vmutils-goos-goarch
 
-release-vmutils-linux-s390x:
-	GOOS=linux GOARCH=s390x $(MAKE) release-vmutils-goos-goarch
-
 release-vmutils-darwin-amd64:
 	GOOS=darwin GOARCH=amd64 $(MAKE) release-vmutils-goos-goarch
 
@@ -373,7 +366,7 @@ release-vmutils-goos-goarch: \
 	vmrestore-$(GOOS)-$(GOARCH)-prod \
 	vmctl-$(GOOS)-$(GOARCH)-prod
 	cd bin && \
-		tar $(TAR_OWNERSHIP) --transform="flags=r;s|-$(GOOS)-$(GOARCH)||" -czf vmutils-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
+		tar --transform="flags=r;s|-$(GOOS)-$(GOARCH)||" -czf vmutils-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
 			vmagent-$(GOOS)-$(GOARCH)-prod \
 			vmalert-$(GOOS)-$(GOARCH)-prod \
 			vmalert-tool-$(GOOS)-$(GOARCH)-prod \
@@ -435,102 +428,71 @@ release-vmutils-windows-goarch: \
 		vmctl-windows-$(GOARCH)-prod.exe
 
 pprof-cpu:
-	go tool pprof -trim_path=github.com/VictoriaMetrics/VictoriaMetrics $(PPROF_FILE)
+	go tool pprof -trim_path=github.com/VictoriaMetrics/VictoriaMetrics@ $(PPROF_FILE)
 
 fmt:
 	gofmt -l -w -s ./lib
 	gofmt -l -w -s ./app
-	gofmt -l -w -s ./apptest
 
 vet:
-	GOEXPERIMENT=synctest go vet ./lib/...
+	go vet ./lib/...
 	go vet ./app/...
-	go vet ./apptest/...
 
 check-all: fmt vet golangci-lint govulncheck
 
-clean-checkers: remove-golangci-lint remove-govulncheck
-
 test:
-	GOEXPERIMENT=synctest go test ./lib/... ./app/...
+	go test ./lib/... ./app/...
 
 test-race:
-	GOEXPERIMENT=synctest go test -race ./lib/... ./app/...
+	go test -race ./lib/... ./app/...
 
 test-pure:
-	GOEXPERIMENT=synctest CGO_ENABLED=0 go test ./lib/... ./app/...
+	CGO_ENABLED=0 go test ./lib/... ./app/...
 
 test-full:
-	GOEXPERIMENT=synctest go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
+	go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
 
 test-full-386:
-	GOEXPERIMENT=synctest GOARCH=386 go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
-
-integration-test:
-	$(MAKE) apptest
-
-apptest:
-	$(MAKE) victoria-metrics vmagent vmalert vmauth vmctl vmbackup vmrestore
-	go test ./apptest/... -skip="^Test(Cluster|Legacy).*"
-
-integration-test-legacy: victoria-metrics vmbackup vmrestore
-	OS=$$(uname | tr '[:upper:]' '[:lower:]'); \
-	ARCH=$$(uname -m | tr '[:upper:]' '[:lower:]' | sed 's/x86_64/amd64/'); \
-	VERSION=v1.132.0; \
-	VMSINGLE=victoria-metrics-$${OS}-$${ARCH}-$${VERSION}.tar.gz; \
-	VMCLUSTER=victoria-metrics-$${OS}-$${ARCH}-$${VERSION}-cluster.tar.gz; \
-	URL=https://github.com/VictoriaMetrics/VictoriaMetrics/releases/download/$${VERSION}; \
-	DIR=/tmp/$${VERSION}; \
-	test -d $${DIR} || (mkdir $${DIR} && \
-		curl --output-dir /tmp -LO $${URL}/$${VMSINGLE} && tar xzf /tmp/$${VMSINGLE} -C $${DIR} && \
-		curl --output-dir /tmp -LO $${URL}/$${VMCLUSTER} && tar xzf /tmp/$${VMCLUSTER} -C $${DIR} \
-	); \
-	VM_LEGACY_VMSINGLE_PATH=$${DIR}/victoria-metrics-prod \
-	VM_LEGACY_VMSTORAGE_PATH=$${DIR}/vmstorage-prod \
-	go test ./apptest/tests -run="^TestLegacySingle.*"
+	GOARCH=386 go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
 
 benchmark:
-	GOEXPERIMENT=synctest go test -bench=. ./lib/...
+	go test -bench=. ./lib/...
 	go test -bench=. ./app/...
 
 benchmark-pure:
-	GOEXPERIMENT=synctest CGO_ENABLED=0 go test -bench=. ./lib/...
+	CGO_ENABLED=0 go test -bench=. ./lib/...
 	CGO_ENABLED=0 go test -bench=. ./app/...
 
 vendor-update:
-	go get -u ./lib/...
-	go get -u ./app/...
-	go mod tidy -compat=1.24
+	go get -u -d ./lib/...
+	go get -u -d ./app/...
+	go mod tidy -compat=1.21
 	go mod vendor
 
 app-local:
-	CGO_ENABLED=1 go build $(RACE) -ldflags "$(GO_BUILDINFO)" -tags "$(EXTRA_GO_BUILD_TAGS)" -o bin/$(APP_NAME)$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
+	CGO_ENABLED=1 go build $(RACE) -ldflags "$(GO_BUILDINFO)" -o bin/$(APP_NAME)$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
 
 app-local-pure:
-	CGO_ENABLED=0 go build $(RACE) -ldflags "$(GO_BUILDINFO)" -tags "$(EXTRA_GO_BUILD_TAGS)" -o bin/$(APP_NAME)-pure$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
+	CGO_ENABLED=0 go build $(RACE) -ldflags "$(GO_BUILDINFO)" -o bin/$(APP_NAME)-pure$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
 
 app-local-goos-goarch:
-	CGO_ENABLED=$(CGO_ENABLED) GOOS=$(GOOS) GOARCH=$(GOARCH) go build $(RACE) -ldflags "$(GO_BUILDINFO)" -tags "$(EXTRA_GO_BUILD_TAGS)" -o bin/$(APP_NAME)-$(GOOS)-$(GOARCH)$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
+	CGO_ENABLED=$(CGO_ENABLED) GOOS=$(GOOS) GOARCH=$(GOARCH) go build $(RACE) -ldflags "$(GO_BUILDINFO)" -o bin/$(APP_NAME)-$(GOOS)-$(GOARCH)$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
 
 app-local-windows-goarch:
-	CGO_ENABLED=0 GOOS=windows GOARCH=$(GOARCH) go build $(RACE) -ldflags "$(GO_BUILDINFO)" -tags "$(EXTRA_GO_BUILD_TAGS)" -o bin/$(APP_NAME)-windows-$(GOARCH)$(RACE).exe $(PKG_PREFIX)/app/$(APP_NAME)
+	CGO_ENABLED=0 GOOS=windows GOARCH=$(GOARCH) go build $(RACE) -ldflags "$(GO_BUILDINFO)" -o bin/$(APP_NAME)-windows-$(GOARCH)$(RACE).exe $(PKG_PREFIX)/app/$(APP_NAME)
 
 quicktemplate-gen: install-qtc
-	qtc -dir=lib
-	qtc -dir=app
+	qtc
 
 install-qtc:
 	which qtc || go install github.com/valyala/quicktemplate/qtc@latest
 
 
 golangci-lint: install-golangci-lint
-	GOEXPERIMENT=synctest golangci-lint run
+	golangci-lint run
 
 install-golangci-lint:
-	which golangci-lint && (golangci-lint --version | grep -q $(GOLANGCI_LINT_VERSION)) || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v$(GOLANGCI_LINT_VERSION)
-
-remove-golangci-lint:
-	rm -rf `which golangci-lint`
+	which golangci-lint || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v1.57.1
 
 govulncheck: install-govulncheck
 	govulncheck ./...
@@ -538,11 +500,38 @@ govulncheck: install-govulncheck
 install-govulncheck:
 	which govulncheck || go install golang.org/x/vuln/cmd/govulncheck@latest
 
-remove-govulncheck:
-	rm -rf `which govulncheck`
-
 install-wwhrd:
 	which wwhrd || go install github.com/frapposelli/wwhrd@latest
 
 check-licenses: install-wwhrd
 	wwhrd check -f .wwhrd.yml
+
+copy-docs:
+# The 'printf' function is used instead of 'echo' or 'echo -e' to handle line breaks (e.g. '\n') in the same way on different operating systems (MacOS/Ubuntu Linux/Arch Linux) and their shells (bash/sh/zsh/fish).
+# For details, see https://github.com/VictoriaMetrics/VictoriaMetrics/pull/4548#issue-1782796419 and https://stackoverflow.com/questions/8467424/echo-newline-in-bash-prints-literal-n
+	echo "---" > ${DST}
+	@if [ ${ORDER} -ne 0 ]; then \
+		echo "sort: ${ORDER}" >> ${DST}; \
+		echo "weight: ${ORDER}" >> ${DST}; \
+		printf "menu:\n  docs:\n    parent: 'victoriametrics'\n    weight: ${ORDER}\n" >> ${DST}; \
+	fi
+
+	echo "title: ${TITLE}" >> ${DST}
+	@if [ ${OLD_URL} ]; then \
+		printf "aliases:\n  - ${OLD_URL}\n" >> ${DST}; \
+	fi
+	echo "---" >> ${DST}
+	cat ${SRC} >> ${DST}
+	sed -i='.tmp' 's/<img src=\"docs\//<img src=\"/' ${DST}
+	rm -rf docs/*.tmp
+
+# Copies docs for all components and adds the order/weight tag, title, menu position and alias with the backward compatible link for the old site.
+# For ORDER=0 it adds no order tag/weight tag.
+# FOR OLD_URL - relative link, used for backward compatibility with the link from documentation based on GitHub pages (old one)
+# FOR OLD_URL='' it adds no alias, it should be empty for every new page, don't change it for already existing links.
+# Images starting with <img src="docs/ are replaced with <img src="
+# Cluster docs are supposed to be ordered as 2nd.
+# The rest of docs is ordered manually.
+docs-sync:
+	SRC=README.md DST=docs/README.md OLD_URL='' ORDER=0 TITLE=VictoriaMetrics $(MAKE) copy-docs
+	SRC=README.md DST=docs/Single-server-VictoriaMetrics.md OLD_URL='/Single-server-VictoriaMetrics.html' TITLE=VictoriaMetrics ORDER=1 $(MAKE) copy-docs

SECURITY.md

@@ -4,14 +4,15 @@
 
 The following versions of VictoriaMetrics receive regular security fixes:
 
-| Version                                                                        | Supported          |
-|--------------------------------------------------------------------------------|--------------------|
-| [Latest release](https://docs.victoriametrics.com/victoriametrics/changelog/)  | :white_check_mark: |
-| [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-releases/) | :white_check_mark: |
-| other releases                                                                 | :x:                |
+| Version | Supported          |
+|---------|--------------------|
+| [latest release](https://docs.victoriametrics.com/CHANGELOG.html) | :white_check_mark: |
+| v1.97.x [LTS line](https://docs.victoriametrics.com/lts-releases/) | :white_check_mark: |
+| v1.93.x [LTS line](https://docs.victoriametrics.com/lts-releases/) | :white_check_mark: |
+| other releases  | :x:                |
 
 See [this page](https://victoriametrics.com/security/) for more details.
 
 ## Reporting a Vulnerability
 
-Please report any security issues to <security@victoriametrics.com>
+Please report any security issues to security@victoriametrics.com

app/victoria-logs/Makefile

@@ -0,0 +1,103 @@
+# All these commands must run from repository root.
+
+victoria-logs:
+	APP_NAME=victoria-logs $(MAKE) app-local
+
+victoria-logs-race:
+	APP_NAME=victoria-logs RACE=-race $(MAKE) app-local
+
+victoria-logs-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker
+
+victoria-logs-pure-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-pure
+
+victoria-logs-linux-amd64-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-linux-amd64
+
+victoria-logs-linux-arm-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-linux-arm
+
+victoria-logs-linux-arm64-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-linux-arm64
+
+victoria-logs-linux-ppc64le-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-linux-ppc64le
+
+victoria-logs-linux-386-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-linux-386
+
+victoria-logs-darwin-amd64-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-darwin-amd64
+
+victoria-logs-darwin-arm64-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-darwin-arm64
+
+victoria-logs-freebsd-amd64-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-freebsd-amd64
+
+victoria-logs-openbsd-amd64-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-openbsd-amd64
+
+victoria-logs-windows-amd64-prod:
+	APP_NAME=victoria-logs $(MAKE) app-via-docker-windows-amd64
+
+package-victoria-logs:
+	APP_NAME=victoria-logs $(MAKE) package-via-docker
+
+package-victoria-logs-pure:
+	APP_NAME=victoria-logs $(MAKE) package-via-docker-pure
+
+package-victoria-logs-amd64:
+	APP_NAME=victoria-logs $(MAKE) package-via-docker-amd64
+
+package-victoria-logs-arm:
+	APP_NAME=victoria-logs $(MAKE) package-via-docker-arm
+
+package-victoria-logs-arm64:
+	APP_NAME=victoria-logs $(MAKE) package-via-docker-arm64
+
+package-victoria-logs-ppc64le:
+	APP_NAME=victoria-logs $(MAKE) package-via-docker-ppc64le
+
+package-victoria-logs-386:
+	APP_NAME=victoria-logs $(MAKE) package-via-docker-386
+
+publish-victoria-logs:
+	APP_NAME=victoria-logs $(MAKE) publish-via-docker
+
+victoria-logs-linux-amd64:
+	APP_NAME=victoria-logs CGO_ENABLED=1 GOOS=linux GOARCH=amd64 $(MAKE) app-local-goos-goarch
+
+victoria-logs-linux-arm:
+	APP_NAME=victoria-logs CGO_ENABLED=0 GOOS=linux GOARCH=arm $(MAKE) app-local-goos-goarch
+
+victoria-logs-linux-arm64:
+	APP_NAME=victoria-logs CGO_ENABLED=0 GOOS=linux GOARCH=arm64 $(MAKE) app-local-goos-goarch
+
+victoria-logs-linux-ppc64le:
+	APP_NAME=victoria-logs CGO_ENABLED=0 GOOS=linux GOARCH=ppc64le $(MAKE) app-local-goos-goarch
+
+victoria-logs-linux-s390x:
+	APP_NAME=victoria-logs CGO_ENABLED=0 GOOS=linux GOARCH=s390x $(MAKE) app-local-goos-goarch
+
+victoria-logs-linux-386:
+	APP_NAME=victoria-logs CGO_ENABLED=0 GOOS=linux GOARCH=386 $(MAKE) app-local-goos-goarch
+
+victoria-logs-darwin-amd64:
+	APP_NAME=victoria-logs CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 $(MAKE) app-local-goos-goarch
+
+victoria-logs-darwin-arm64:
+	APP_NAME=victoria-logs CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 $(MAKE) app-local-goos-goarch
+
+victoria-logs-freebsd-amd64:
+	APP_NAME=victoria-logs CGO_ENABLED=0 GOOS=freebsd GOARCH=amd64 $(MAKE) app-local-goos-goarch
+
+victoria-logs-openbsd-amd64:
+	APP_NAME=victoria-logs CGO_ENABLED=0 GOOS=openbsd GOARCH=amd64 $(MAKE) app-local-goos-goarch
+
+victoria-logs-windows-amd64:
+	GOARCH=amd64 APP_NAME=victoria-logs $(MAKE) app-local-windows-goarch
+
+victoria-logs-pure:
+	APP_NAME=victoria-logs $(MAKE) app-local-pure

app/victoria-logs/README.md

@@ -1 +0,0 @@
-VictoriaLogs source code has been moved to [github.com/VictoriaMetrics/VictoriaLogs](https://github.com/VictoriaMetrics/VictoriaLogs/).

app/victoria-logs/deployment/Dockerfile

@@ -0,0 +1,8 @@
+ARG base_image
+FROM $base_image
+
+EXPOSE 9428
+
+ENTRYPOINT ["/victoria-logs-prod"]
+ARG src_binary
+COPY $src_binary ./victoria-logs-prod

app/victoria-logs/main.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlselect"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
+)
+
+var (
+	httpListenAddrs  = flagutil.NewArrayString("httpListenAddr", "TCP address to listen for incoming http requests. See also -httpListenAddr.useProxyProtocol")
+	useProxyProtocol = flagutil.NewArrayBool("httpListenAddr.useProxyProtocol", "Whether to use proxy protocol for connections accepted at the given -httpListenAddr . "+
+		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt . "+
+		"With enabled proxy protocol http server cannot serve regular /metrics endpoint. Use -pushmetrics.url for metrics pushing")
+)
+
+func main() {
+	// Write flags and help message to stdout, since it is easier to grep or pipe.
+	flag.CommandLine.SetOutput(os.Stdout)
+	flag.Usage = usage
+	envflag.Parse()
+	buildinfo.Init()
+	logger.Init()
+
+	listenAddrs := *httpListenAddrs
+	if len(listenAddrs) == 0 {
+		listenAddrs = []string{":9428"}
+	}
+	logger.Infof("starting VictoriaLogs at %q...", listenAddrs)
+	startTime := time.Now()
+
+	vlstorage.Init()
+	vlselect.Init()
+	vlinsert.Init()
+
+	go httpserver.Serve(listenAddrs, useProxyProtocol, requestHandler)
+	logger.Infof("started VictoriaLogs in %.3f seconds; see https://docs.victoriametrics.com/VictoriaLogs/", time.Since(startTime).Seconds())
+
+	pushmetrics.Init()
+	sig := procutil.WaitForSigterm()
+	logger.Infof("received signal %s", sig)
+	pushmetrics.Stop()
+
+	logger.Infof("gracefully shutting down webservice at %q", listenAddrs)
+	startTime = time.Now()
+	if err := httpserver.Stop(listenAddrs); err != nil {
+		logger.Fatalf("cannot stop the webservice: %s", err)
+	}
+	logger.Infof("successfully shut down the webservice in %.3f seconds", time.Since(startTime).Seconds())
+
+	vlinsert.Stop()
+	vlselect.Stop()
+	vlstorage.Stop()
+
+	fs.MustStopDirRemover()
+
+	logger.Infof("the VictoriaLogs has been stopped in %.3f seconds", time.Since(startTime).Seconds())
+}
+
+func requestHandler(w http.ResponseWriter, r *http.Request) bool {
+	if r.URL.Path == "/" {
+		if r.Method != http.MethodGet {
+			return false
+		}
+		w.Header().Add("Content-Type", "text/html; charset=utf-8")
+		fmt.Fprintf(w, "<h2>Single-node VictoriaLogs</h2></br>")
+		fmt.Fprintf(w, "See docs at <a href='https://docs.victoriametrics.com/VictoriaLogs/'>https://docs.victoriametrics.com/VictoriaLogs/</a></br>")
+		fmt.Fprintf(w, "Useful endpoints:</br>")
+		httpserver.WriteAPIHelp(w, [][2]string{
+			{"select/vmui", "Web UI for VictoriaLogs"},
+			{"metrics", "available service metrics"},
+			{"flags", "command-line flags"},
+		})
+		return true
+	}
+	if vlinsert.RequestHandler(w, r) {
+		return true
+	}
+	if vlselect.RequestHandler(w, r) {
+		return true
+	}
+	return false
+}
+
+func usage() {
+	const s = `
+victoria-logs is a log management and analytics service.
+
+See the docs at https://docs.victoriametrics.com/VictoriaLogs/
+`
+	flagutil.Usage(s)
+}

app/victoria-logs/multiarch/Dockerfile

@@ -0,0 +1,12 @@
+# See https://medium.com/on-docker/use-multi-stage-builds-to-inject-ca-certs-ad1e8f01de1b
+ARG certs_image
+ARG root_image
+FROM $certs_image as certs
+RUN apk update && apk upgrade && apk --update --no-cache add ca-certificates
+
+FROM $root_image
+COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
+EXPOSE 9428
+ENTRYPOINT ["/victoria-logs-prod"]
+ARG TARGETARCH
+COPY victoria-logs-linux-${TARGETARCH}-prod ./victoria-logs-prod

app/victoria-metrics/Makefile

@@ -27,9 +27,6 @@ victoria-metrics-linux-ppc64le-prod:
 victoria-metrics-linux-386-prod:
 	APP_NAME=victoria-metrics $(MAKE) app-via-docker-linux-386
 
-victoria-metrics-linux-s390x-prod:
-	APP_NAME=victoria-metrics $(MAKE) app-via-docker-linux-s390x
-
 victoria-metrics-darwin-amd64-prod:
 	APP_NAME=victoria-metrics $(MAKE) app-via-docker-darwin-amd64
 
@@ -91,9 +88,6 @@ victoria-metrics-linux-ppc64le:
 victoria-metrics-linux-s390x:
 	APP_NAME=victoria-metrics CGO_ENABLED=0 GOOS=linux GOARCH=s390x $(MAKE) app-local-goos-goarch
 
-victoria-metrics-linux-loong64:
-	APP_NAME=victoria-metrics CGO_ENABLED=0 GOOS=linux GOARCH=loong64 $(MAKE) app-local-goos-goarch
-
 victoria-metrics-linux-386:
 	APP_NAME=victoria-metrics CGO_ENABLED=0 GOOS=linux GOARCH=386 $(MAKE) app-local-goos-goarch
 

app/victoria-metrics/deployment/Dockerfile

@@ -1,8 +1,8 @@
-ARG base_image=non-existing
+ARG base_image
 FROM $base_image
 
 EXPOSE 8428
 
 ENTRYPOINT ["/victoria-metrics-prod"]
-ARG src_binary=non-existing
+ARG src_binary
 COPY $src_binary ./victoria-metrics-prod

app/victoria-metrics/main.go

@@ -14,9 +14,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
@@ -31,7 +31,7 @@ var (
 		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt . "+
 		"With enabled proxy protocol http server cannot serve regular /metrics endpoint. Use -pushmetrics.url for metrics pushing")
 	minScrapeInterval = flag.Duration("dedup.minScrapeInterval", 0, "Leave only the last sample in every time series per each discrete interval "+
-		"equal to -dedup.minScrapeInterval > 0. See also -streamAggr.dedupInterval and https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#deduplication")
+		"equal to -dedup.minScrapeInterval > 0. See also -streamAggr.dedupInterval and https://docs.victoriametrics.com/#deduplication")
 	dryRun = flag.Bool("dryRun", false, "Whether to check config files without running VictoriaMetrics. The following config files are checked: "+
 		"-promscrape.config, -relabelConfig and -streamAggr.config. Unknown config entries aren't allowed in -promscrape.config by default. "+
 		"This can be changed with -promscrape.config.strictParse=false command-line flag")
@@ -39,24 +39,9 @@ var (
 		"The saved data survives unclean shutdowns such as OOM crash, hardware reset, SIGKILL, etc. "+
 		"Bigger intervals may help increase the lifetime of flash storage with limited write cycles (e.g. Raspberry PI). "+
 		"Smaller intervals increase disk IO load. Minimum supported value is 1s")
-	maxIngestionRate = flag.Int("maxIngestionRate", 0, "The maximum number of samples vmsingle can receive per second. Data ingestion is paused when the limit is exceeded. "+
-		"By default there are no limits on samples ingestion rate.")
-	finalDedupScheduleInterval = flag.Duration("storage.finalDedupScheduleCheckInterval", time.Hour, "The interval for checking when final deduplication process should be started."+
-		"Storage unconditionally adds 25% jitter to the interval value on each check evaluation."+
-		" Changing the interval to the bigger values may delay downsampling, deduplication for historical data."+
-		" See also https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#deduplication")
 )
 
 func main() {
-	// VictoriaMetrics is optimized for reduced memory allocations,
-	// so it can run with the reduced GOGC in order to reduce the used memory,
-	// while keeping CPU usage spent in GC at low levels.
-	//
-	// Some workloads may need increased GOGC values. Then such values can be set via GOGC environment variable.
-	// It is recommended increasing GOGC if go_memstats_gc_cpu_fraction metric exposed at /metrics page
-	// exceeds 0.05 for extended periods of time.
-	cgroup.SetGOGC(30)
-
 	// Write flags and help message to stdout, since it is easier to grep or pipe.
 	flag.CommandLine.SetOutput(os.Stdout)
 	flag.Usage = usage
@@ -89,20 +74,13 @@ func main() {
 	startTime := time.Now()
 	storage.SetDedupInterval(*minScrapeInterval)
 	storage.SetDataFlushInterval(*inmemoryDataFlushInterval)
-	if *finalDedupScheduleInterval < time.Hour {
-		logger.Fatalf("-dedup.finalDedupScheduleCheckInterval cannot be smaller than 1 hour; got %s", *finalDedupScheduleInterval)
-	}
-	storage.SetFinalDedupScheduleInterval(*finalDedupScheduleInterval)
 	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
 	vmselect.Init()
-	vminsertcommon.StartIngestionRateLimiter(*maxIngestionRate)
 	vminsert.Init()
 
 	startSelfScraper()
 
-	go httpserver.Serve(listenAddrs, requestHandler, httpserver.ServeOptions{
-		UseProxyProtocol: useProxyProtocol,
-	})
+	go httpserver.Serve(listenAddrs, useProxyProtocol, requestHandler)
 	logger.Infof("started VictoriaMetrics in %.3f seconds", time.Since(startTime).Seconds())
 
 	pushmetrics.Init()
@@ -119,11 +97,12 @@ func main() {
 	}
 	logger.Infof("successfully shut down the webservice in %.3f seconds", time.Since(startTime).Seconds())
 	vminsert.Stop()
-	vminsertcommon.StopIngestionRateLimiter()
 
 	vmstorage.Stop()
 	vmselect.Stop()
 
+	fs.MustStopDirRemover()
+
 	logger.Infof("the VictoriaMetrics has been stopped in %.3f seconds", time.Since(startTime).Seconds())
 }
 
@@ -134,7 +113,6 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		}
 		w.Header().Add("Content-Type", "text/html; charset=utf-8")
 		fmt.Fprintf(w, "<h2>Single-node VictoriaMetrics</h2></br>")
-		fmt.Fprintf(w, "Version %s<br>", buildinfo.Version)
 		fmt.Fprintf(w, "See docs at <a href='https://docs.victoriametrics.com/'>https://docs.victoriametrics.com/</a></br>")
 		fmt.Fprintf(w, "Useful endpoints:</br>")
 		httpserver.WriteAPIHelp(w, [][2]string{
@@ -170,7 +148,7 @@ func usage() {
 	const s = `
 victoria-metrics is a time series database and monitoring solution.
 
-See the docs at https://docs.victoriametrics.com/victoriametrics/
+See the docs at https://docs.victoriametrics.com/
 `
 	flagutil.Usage(s)
 }

app/victoria-metrics/main_test.go

@@ -0,0 +1,602 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"log"
+	"math/rand"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"reflect"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	testutil "github.com/VictoriaMetrics/VictoriaMetrics/app/victoria-metrics/test"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+)
+
+const (
+	testFixturesDir            = "testdata"
+	testStorageSuffix          = "vm-test-storage"
+	testHTTPListenAddr         = ":7654"
+	testStatsDListenAddr       = ":2003"
+	testOpenTSDBListenAddr     = ":4242"
+	testOpenTSDBHTTPListenAddr = ":4243"
+	testLogLevel               = "INFO"
+)
+
+const (
+	testReadHTTPPath           = "http://127.0.0.1" + testHTTPListenAddr
+	testWriteHTTPPath          = "http://127.0.0.1" + testHTTPListenAddr + "/write"
+	testOpenTSDBWriteHTTPPath  = "http://127.0.0.1" + testOpenTSDBHTTPListenAddr + "/api/put"
+	testPromWriteHTTPPath      = "http://127.0.0.1" + testHTTPListenAddr + "/api/v1/write"
+	testImportCSVWriteHTTPPath = "http://127.0.0.1" + testHTTPListenAddr + "/api/v1/import/csv"
+
+	testHealthHTTPPath = "http://127.0.0.1" + testHTTPListenAddr + "/health"
+)
+
+const (
+	testStorageInitTimeout = 10 * time.Second
+)
+
+var (
+	storagePath   string
+	insertionTime = time.Now().UTC()
+)
+
+type test struct {
+	Name                     string   `json:"name"`
+	Data                     []string `json:"data"`
+	InsertQuery              string   `json:"insert_query"`
+	Query                    []string `json:"query"`
+	ResultMetrics            []Metric `json:"result_metrics"`
+	ResultSeries             Series   `json:"result_series"`
+	ResultQuery              Query    `json:"result_query"`
+	Issue                    string   `json:"issue"`
+	ExpectedResultLinesCount int      `json:"expected_result_lines_count"`
+}
+
+type Metric struct {
+	Metric     map[string]string `json:"metric"`
+	Values     []float64         `json:"values"`
+	Timestamps []int64           `json:"timestamps"`
+}
+
+func (r *Metric) UnmarshalJSON(b []byte) error {
+	type plain Metric
+	return json.Unmarshal(testutil.PopulateTimeTpl(b, insertionTime), (*plain)(r))
+}
+
+type Series struct {
+	Status string              `json:"status"`
+	Data   []map[string]string `json:"data"`
+}
+
+type Query struct {
+	Status string `json:"status"`
+	Data   struct {
+		ResultType string          `json:"resultType"`
+		Result     json.RawMessage `json:"result"`
+	} `json:"data"`
+}
+
+const rtVector, rtMatrix = "vector", "matrix"
+
+func (q *Query) metrics() ([]Metric, error) {
+	switch q.Data.ResultType {
+	case rtVector:
+		var r QueryInstant
+		if err := json.Unmarshal(q.Data.Result, &r.Result); err != nil {
+			return nil, err
+		}
+		return r.metrics()
+	case rtMatrix:
+		var r QueryRange
+		if err := json.Unmarshal(q.Data.Result, &r.Result); err != nil {
+			return nil, err
+		}
+		return r.metrics()
+	default:
+		return nil, fmt.Errorf("unknown result type %q", q.Data.ResultType)
+	}
+}
+
+type QueryInstant struct {
+	Result []struct {
+		Labels map[string]string `json:"metric"`
+		TV     [2]interface{}    `json:"value"`
+	} `json:"result"`
+}
+
+func (q QueryInstant) metrics() ([]Metric, error) {
+	result := make([]Metric, len(q.Result))
+	for i, res := range q.Result {
+		f, err := strconv.ParseFloat(res.TV[1].(string), 64)
+		if err != nil {
+			return nil, fmt.Errorf("metric %v, unable to parse float64 from %s: %w", res, res.TV[1], err)
+		}
+		var m Metric
+		m.Metric = res.Labels
+		m.Timestamps = append(m.Timestamps, int64(res.TV[0].(float64)))
+		m.Values = append(m.Values, f)
+		result[i] = m
+	}
+	return result, nil
+}
+
+type QueryRange struct {
+	Result []struct {
+		Metric map[string]string `json:"metric"`
+		Values [][]interface{}   `json:"values"`
+	} `json:"result"`
+}
+
+func (q QueryRange) metrics() ([]Metric, error) {
+	var result []Metric
+	for i, res := range q.Result {
+		var m Metric
+		for _, tv := range res.Values {
+			f, err := strconv.ParseFloat(tv[1].(string), 64)
+			if err != nil {
+				return nil, fmt.Errorf("metric %v, unable to parse float64 from %s: %w", res, tv[1], err)
+			}
+			m.Values = append(m.Values, f)
+			m.Timestamps = append(m.Timestamps, int64(tv[0].(float64)))
+		}
+		if len(m.Values) < 1 || len(m.Timestamps) < 1 {
+			return nil, fmt.Errorf("metric %v contains no values", res)
+		}
+		m.Metric = q.Result[i].Metric
+		result = append(result, m)
+	}
+	return result, nil
+}
+
+func (q *Query) UnmarshalJSON(b []byte) error {
+	type plain Query
+	return json.Unmarshal(testutil.PopulateTimeTpl(b, insertionTime), (*plain)(q))
+}
+
+func TestMain(m *testing.M) {
+	setUp()
+	code := m.Run()
+	tearDown()
+	os.Exit(code)
+}
+
+func setUp() {
+	storagePath = filepath.Join(os.TempDir(), testStorageSuffix)
+	processFlags()
+	logger.Init()
+	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
+	vmselect.Init()
+	vminsert.Init()
+	go httpserver.Serve(*httpListenAddrs, useProxyProtocol, requestHandler)
+	readyStorageCheckFunc := func() bool {
+		resp, err := http.Get(testHealthHTTPPath)
+		if err != nil {
+			return false
+		}
+		_ = resp.Body.Close()
+		return resp.StatusCode == 200
+	}
+	if err := waitFor(testStorageInitTimeout, readyStorageCheckFunc); err != nil {
+		log.Fatalf("http server can't start for %s seconds, err %s", testStorageInitTimeout, err)
+	}
+}
+
+func processFlags() {
+	flag.Parse()
+	for _, fv := range []struct {
+		flag  string
+		value string
+	}{
+		{flag: "storageDataPath", value: storagePath},
+		{flag: "httpListenAddr", value: testHTTPListenAddr},
+		{flag: "graphiteListenAddr", value: testStatsDListenAddr},
+		{flag: "opentsdbListenAddr", value: testOpenTSDBListenAddr},
+		{flag: "loggerLevel", value: testLogLevel},
+		{flag: "opentsdbHTTPListenAddr", value: testOpenTSDBHTTPListenAddr},
+	} {
+		// panics if flag doesn't exist
+		if err := flag.Lookup(fv.flag).Value.Set(fv.value); err != nil {
+			log.Fatalf("unable to set %q with value %q, err: %v", fv.flag, fv.value, err)
+		}
+	}
+}
+
+func waitFor(timeout time.Duration, f func() bool) error {
+	fraction := timeout / 10
+	for i := fraction; i < timeout; i += fraction {
+		if f() {
+			return nil
+		}
+		time.Sleep(fraction)
+	}
+	return fmt.Errorf("timeout")
+}
+
+func tearDown() {
+	if err := httpserver.Stop(*httpListenAddrs); err != nil {
+		log.Printf("cannot stop the webservice: %s", err)
+	}
+	vminsert.Stop()
+	vmstorage.Stop()
+	vmselect.Stop()
+	fs.MustRemoveAll(storagePath)
+}
+
+func TestWriteRead(t *testing.T) {
+	t.Run("write", testWrite)
+	time.Sleep(500 * time.Millisecond)
+	vmstorage.Storage.DebugFlush()
+	time.Sleep(1500 * time.Millisecond)
+	t.Run("read", testRead)
+}
+
+func testWrite(t *testing.T) {
+	t.Run("prometheus", func(t *testing.T) {
+		for _, test := range readIn("prometheus", t, insertionTime) {
+			if test.Data == nil {
+				continue
+			}
+			s := newSuite(t)
+			r := testutil.WriteRequest{}
+			s.noError(json.Unmarshal([]byte(strings.Join(test.Data, "\n")), &r.Timeseries))
+			data, err := testutil.Compress(r)
+			s.greaterThan(len(r.Timeseries), 0)
+			if err != nil {
+				t.Errorf("error compressing %v %s", r, err)
+				t.Fail()
+			}
+			httpWrite(t, testPromWriteHTTPPath, test.InsertQuery, bytes.NewBuffer(data))
+		}
+	})
+	t.Run("csv", func(t *testing.T) {
+		for _, test := range readIn("csv", t, insertionTime) {
+			if test.Data == nil {
+				continue
+			}
+			httpWrite(t, testImportCSVWriteHTTPPath, test.InsertQuery, bytes.NewBuffer([]byte(strings.Join(test.Data, "\n"))))
+		}
+	})
+
+	t.Run("influxdb", func(t *testing.T) {
+		for _, x := range readIn("influxdb", t, insertionTime) {
+			test := x
+			t.Run(test.Name, func(t *testing.T) {
+				t.Parallel()
+				httpWrite(t, testWriteHTTPPath, test.InsertQuery, bytes.NewBufferString(strings.Join(test.Data, "\n")))
+			})
+		}
+	})
+	t.Run("graphite", func(t *testing.T) {
+		for _, x := range readIn("graphite", t, insertionTime) {
+			test := x
+			t.Run(test.Name, func(t *testing.T) {
+				t.Parallel()
+				tcpWrite(t, "127.0.0.1"+testStatsDListenAddr, strings.Join(test.Data, "\n"))
+			})
+		}
+	})
+	t.Run("opentsdb", func(t *testing.T) {
+		for _, x := range readIn("opentsdb", t, insertionTime) {
+			test := x
+			t.Run(test.Name, func(t *testing.T) {
+				t.Parallel()
+				tcpWrite(t, "127.0.0.1"+testOpenTSDBListenAddr, strings.Join(test.Data, "\n"))
+			})
+		}
+	})
+	t.Run("opentsdbhttp", func(t *testing.T) {
+		for _, x := range readIn("opentsdbhttp", t, insertionTime) {
+			test := x
+			t.Run(test.Name, func(t *testing.T) {
+				t.Parallel()
+				logger.Infof("writing %s", test.Data)
+				httpWrite(t, testOpenTSDBWriteHTTPPath, test.InsertQuery, bytes.NewBufferString(strings.Join(test.Data, "\n")))
+			})
+		}
+	})
+}
+
+func testRead(t *testing.T) {
+	for _, engine := range []string{"csv", "prometheus", "graphite", "opentsdb", "influxdb", "opentsdbhttp"} {
+		t.Run(engine, func(t *testing.T) {
+			for _, x := range readIn(engine, t, insertionTime) {
+				test := x
+				t.Run(test.Name, func(t *testing.T) {
+					t.Parallel()
+					for _, q := range test.Query {
+						q = testutil.PopulateTimeTplString(q, insertionTime)
+						if test.Issue != "" {
+							test.Issue = "\nRegression in " + test.Issue
+						}
+						switch {
+						case strings.HasPrefix(q, "/api/v1/export/csv"):
+							data := strings.Split(string(httpReadData(t, testReadHTTPPath, q)), "\n")
+							if len(data) == test.ExpectedResultLinesCount {
+								t.Fatalf("not expected number of csv lines want=%d\ngot=%d test=%s.%s\n\response=%q", len(data), test.ExpectedResultLinesCount, q, test.Issue, strings.Join(data, "\n"))
+							}
+						case strings.HasPrefix(q, "/api/v1/export"):
+							if err := checkMetricsResult(httpReadMetrics(t, testReadHTTPPath, q), test.ResultMetrics); err != nil {
+								t.Fatalf("Export. %s fails with error %s.%s", q, err, test.Issue)
+							}
+						case strings.HasPrefix(q, "/api/v1/series"):
+							s := Series{}
+							httpReadStruct(t, testReadHTTPPath, q, &s)
+							if err := checkSeriesResult(s, test.ResultSeries); err != nil {
+								t.Fatalf("Series. %s fails with error %s.%s", q, err, test.Issue)
+							}
+						case strings.HasPrefix(q, "/api/v1/query"):
+							queryResult := Query{}
+							httpReadStruct(t, testReadHTTPPath, q, &queryResult)
+							gotMetrics, err := queryResult.metrics()
+							if err != nil {
+								t.Fatalf("failed to parse query response: %s", err)
+							}
+							expMetrics, err := test.ResultQuery.metrics()
+							if err != nil {
+								t.Fatalf("failed to parse expected response: %s", err)
+							}
+							if err := checkMetricsResult(gotMetrics, expMetrics); err != nil {
+								t.Fatalf("%q fails with error %s.%s", q, err, test.Issue)
+							}
+						default:
+							t.Fatalf("unsupported read query %s", q)
+						}
+					}
+				})
+			}
+		})
+	}
+}
+
+func readIn(readFor string, t *testing.T, insertTime time.Time) []test {
+	t.Helper()
+	s := newSuite(t)
+	var tt []test
+	s.noError(filepath.Walk(filepath.Join(testFixturesDir, readFor), func(path string, _ os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if filepath.Ext(path) != ".json" {
+			return nil
+		}
+		b, err := os.ReadFile(path)
+		s.noError(err)
+		item := test{}
+		s.noError(json.Unmarshal(b, &item))
+		for i := range item.Data {
+			item.Data[i] = testutil.PopulateTimeTplString(item.Data[i], insertTime)
+		}
+		tt = append(tt, item)
+		return nil
+	}))
+	if len(tt) == 0 {
+		t.Fatalf("no test found in %s", filepath.Join(testFixturesDir, readFor))
+	}
+	return tt
+}
+
+func httpWrite(t *testing.T, address, query string, r io.Reader) {
+	t.Helper()
+	s := newSuite(t)
+	resp, err := http.Post(address+query, "", r)
+	s.noError(err)
+	s.noError(resp.Body.Close())
+	s.equalInt(resp.StatusCode, 204)
+}
+
+func tcpWrite(t *testing.T, address string, data string) {
+	t.Helper()
+	s := newSuite(t)
+	conn, err := net.Dial("tcp", address)
+	s.noError(err)
+	defer func() {
+		_ = conn.Close()
+	}()
+	n, err := conn.Write([]byte(data))
+	s.noError(err)
+	s.equalInt(n, len(data))
+}
+
+func httpReadMetrics(t *testing.T, address, query string) []Metric {
+	t.Helper()
+	s := newSuite(t)
+	resp, err := http.Get(address + query)
+	s.noError(err)
+	defer func() {
+		_ = resp.Body.Close()
+	}()
+	s.equalInt(resp.StatusCode, 200)
+	var rows []Metric
+	for dec := json.NewDecoder(resp.Body); dec.More(); {
+		var row Metric
+		s.noError(dec.Decode(&row))
+		rows = append(rows, row)
+	}
+	return rows
+}
+
+func httpReadStruct(t *testing.T, address, query string, dst interface{}) {
+	t.Helper()
+	s := newSuite(t)
+	resp, err := http.Get(address + query)
+	s.noError(err)
+	defer func() {
+		_ = resp.Body.Close()
+	}()
+	s.equalInt(resp.StatusCode, 200)
+	s.noError(json.NewDecoder(resp.Body).Decode(dst))
+}
+
+func httpReadData(t *testing.T, address, query string) []byte {
+	t.Helper()
+	s := newSuite(t)
+	resp, err := http.Get(address + query)
+	s.noError(err)
+	defer func() {
+		_ = resp.Body.Close()
+	}()
+	s.equalInt(resp.StatusCode, 200)
+	data, err := io.ReadAll(resp.Body)
+	s.noError(err)
+	return data
+}
+
+func checkMetricsResult(got, want []Metric) error {
+	for _, r := range append([]Metric(nil), got...) {
+		want = removeIfFoundMetrics(r, want)
+	}
+	if len(want) > 0 {
+		return fmt.Errorf("expected metrics %+v not found in %+v", want, got)
+	}
+	return nil
+}
+
+func removeIfFoundMetrics(r Metric, contains []Metric) []Metric {
+	for i, item := range contains {
+		if reflect.DeepEqual(r.Metric, item.Metric) && reflect.DeepEqual(r.Values, item.Values) &&
+			reflect.DeepEqual(r.Timestamps, item.Timestamps) {
+			contains[i] = contains[len(contains)-1]
+			return contains[:len(contains)-1]
+		}
+	}
+	return contains
+}
+
+func checkSeriesResult(got, want Series) error {
+	if got.Status != want.Status {
+		return fmt.Errorf("status mismatch %q - %q", want.Status, got.Status)
+	}
+	wantData := append([]map[string]string(nil), want.Data...)
+	for _, r := range got.Data {
+		wantData = removeIfFoundSeries(r, wantData)
+	}
+	if len(wantData) > 0 {
+		return fmt.Errorf("expected seria(s) %+v not found in %+v", wantData, got.Data)
+	}
+	return nil
+}
+
+func removeIfFoundSeries(r map[string]string, contains []map[string]string) []map[string]string {
+	for i, item := range contains {
+		if reflect.DeepEqual(r, item) {
+			contains[i] = contains[len(contains)-1]
+			return contains[:len(contains)-1]
+		}
+	}
+	return contains
+}
+
+type suite struct{ t *testing.T }
+
+func newSuite(t *testing.T) *suite { return &suite{t: t} }
+
+func (s *suite) noError(err error) {
+	s.t.Helper()
+	if err != nil {
+		s.t.Errorf("unexpected error %v", err)
+		s.t.FailNow()
+	}
+}
+
+func (s *suite) equalInt(a, b int) {
+	s.t.Helper()
+	if a != b {
+		s.t.Errorf("%d not equal %d", a, b)
+		s.t.FailNow()
+	}
+}
+
+func (s *suite) greaterThan(a, b int) {
+	s.t.Helper()
+	if a <= b {
+		s.t.Errorf("%d less or equal then %d", a, b)
+		s.t.FailNow()
+	}
+}
+
+func TestImportJSONLines(t *testing.T) {
+	f := func(labelsCount, labelLen int) {
+		t.Helper()
+
+		reqURL := fmt.Sprintf("http://localhost%s/api/v1/import", testHTTPListenAddr)
+		line := generateJSONLine(labelsCount, labelLen)
+		req, err := http.NewRequest("POST", reqURL, bytes.NewBufferString(line))
+		if err != nil {
+			t.Fatalf("cannot create request: %s", err)
+		}
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			t.Fatalf("cannot perform request for labelsCount=%d, labelLen=%d: %s", labelsCount, labelLen, err)
+		}
+		if resp.StatusCode != 204 {
+			t.Fatalf("unexpected statusCode for labelsCount=%d, labelLen=%d; got %d; want 204", labelsCount, labelLen, resp.StatusCode)
+		}
+	}
+
+	// labels with various lengths
+	for i := 0; i < 500; i++ {
+		f(10, i*5)
+	}
+
+	// Too many labels
+	f(1000, 100)
+
+	// Too long labels
+	f(1, 100_000)
+	f(10, 100_000)
+	f(10, 10_000)
+}
+
+func generateJSONLine(labelsCount, labelLen int) string {
+	m := make(map[string]string, labelsCount)
+	m["__name__"] = generateSizedRandomString(labelLen)
+	for j := 1; j < labelsCount; j++ {
+		labelName := generateSizedRandomString(labelLen)
+		labelValue := generateSizedRandomString(labelLen)
+		m[labelName] = labelValue
+	}
+
+	type jsonLine struct {
+		Metric     map[string]string `json:"metric"`
+		Values     []float64         `json:"values"`
+		Timestamps []int64           `json:"timestamps"`
+	}
+	line := &jsonLine{
+		Metric:     m,
+		Values:     []float64{1.34},
+		Timestamps: []int64{time.Now().UnixNano() / 1e6},
+	}
+	data, err := json.Marshal(&line)
+	if err != nil {
+		panic(fmt.Errorf("cannot marshal JSON: %w", err))
+	}
+	data = append(data, '\n')
+	return string(data)
+}
+
+const alphabetSample = `qwertyuiopasdfghjklzxcvbnm`
+
+func generateSizedRandomString(size int) string {
+	dst := make([]byte, size)
+	for i := range dst {
+		dst[i] = alphabetSample[rand.Intn(len(alphabetSample))]
+	}
+	return string(dst)
+}

app/victoria-metrics/multiarch/Dockerfile

@@ -1,7 +1,7 @@
 # See https://medium.com/on-docker/use-multi-stage-builds-to-inject-ca-certs-ad1e8f01de1b
-ARG certs_image=non-existing
-ARG root_image=non-existing
-FROM $certs_image AS certs
+ARG certs_image
+ARG root_image
+FROM $certs_image as certs
 RUN apk update && apk upgrade && apk --update --no-cache add ca-certificates
 
 FROM $root_image
@@ -9,5 +9,4 @@ COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certifica
 EXPOSE 8428
 ENTRYPOINT ["/victoria-metrics-prod"]
 ARG TARGETARCH
-ARG BINARY_SUFFIX=non-existing
-COPY victoria-metrics-linux-${TARGETARCH}-prod${BINARY_SUFFIX} ./victoria-metrics-prod
+COPY victoria-metrics-linux-${TARGETARCH}-prod ./victoria-metrics-prod

app/victoria-metrics/self_scraper.go

@@ -10,12 +10,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prommetadata"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricsmetadata"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
 )
 
 var (
@@ -29,9 +26,11 @@ var selfScraperWG sync.WaitGroup
 
 func startSelfScraper() {
 	selfScraperStopCh = make(chan struct{})
-	selfScraperWG.Go(func() {
+	selfScraperWG.Add(1)
+	go func() {
+		defer selfScraperWG.Done()
 		selfScraper(*selfScrapeInterval)
-	})
+	}()
 }
 
 func stopSelfScraper() {
@@ -48,7 +47,6 @@ func selfScraper(scrapeInterval time.Duration) {
 
 	var bb bytesutil.ByteBuffer
 	var rows prometheus.Rows
-	var metadataRows prometheus.MetadataRows
 	var mrs []storage.MetricRow
 	var labels []prompb.Label
 	t := time.NewTicker(scrapeInterval)
@@ -58,12 +56,7 @@ func selfScraper(scrapeInterval time.Duration) {
 		appmetrics.WritePrometheusMetrics(&bb)
 		s := bytesutil.ToUnsafeString(bb.B)
 		rows.Reset()
-		// Parse metrics and optionally metadata when enabled
-		if prommetadata.IsEnabled() {
-			rows, metadataRows = prometheus.UnmarshalWithMetadata(rows, metadataRows, s, nil)
-		} else {
-			rows.UnmarshalWithErrLogger(s, nil)
-		}
+		rows.Unmarshal(s)
 		mrs = mrs[:0]
 		for i := range rows.Rows {
 			r := &rows.Rows[i]
@@ -75,10 +68,6 @@ func selfScraper(scrapeInterval time.Duration) {
 				t := &r.Tags[j]
 				labels = addLabel(labels, t.Key, t.Value)
 			}
-			if timeserieslimits.IsExceeding(labels) {
-				// Skip metric with exceeding labels.
-				continue
-			}
 			if len(mrs) < cap(mrs) {
 				mrs = mrs[:len(mrs)+1]
 			} else {
@@ -96,19 +85,6 @@ func selfScraper(scrapeInterval time.Duration) {
 		if err := vmstorage.AddRows(mrs); err != nil {
 			logger.Errorf("cannot store self-scraped metrics: %s", err)
 		}
-		if len(metadataRows.Rows) > 0 {
-			mms := make([]metricsmetadata.Row, 0, len(metadataRows.Rows))
-			for _, mm := range metadataRows.Rows {
-				mms = append(mms, metricsmetadata.Row{
-					MetricFamilyName: bytesutil.ToUnsafeBytes(mm.Metric),
-					Help:             bytesutil.ToUnsafeBytes(mm.Help),
-					Type:             mm.Type,
-				})
-			}
-			if err := vmstorage.AddMetadataRows(mms); err != nil {
-				logger.Errorf("cannot store self-scraped metrics metadata: %s", err)
-			}
-		}
 	}
 	for {
 		select {

app/victoria-metrics/test/prom_writer.go

@@ -1,16 +1,12 @@
 package test
 
-import (
-	"fmt"
-
-	"github.com/golang/snappy"
-)
+import "github.com/golang/snappy"
 
 // Compress marshals and compresses wr.
-func Compress(wr WriteRequest) []byte {
+func Compress(wr WriteRequest) ([]byte, error) {
 	data, err := wr.Marshal()
 	if err != nil {
-		panic(fmt.Errorf("BUG: cannot compress WriteRequest: %s", err))
+		return nil, err
 	}
-	return snappy.Encode(nil, data)
+	return snappy.Encode(nil, data), nil
 }

app/victoria-metrics/testdata/csv/basic.json

@@ -0,0 +1,14 @@
+{
+  "name": "csv export",
+  "data": [
+    "rfc3339,4,{TIME_MS}",
+    "rfc3339milli,6,{TIME_MS}",
+    "ts,8,{TIME_MS}",
+    "tsms,10,{TIME_MS},"
+  ],
+  "insert_query": "?format=1:label:tfmt,2:metric:test_csv,3:time:unix_ms",
+  "query": [
+    "/api/v1/export/csv?format=__name__,tfmt,__value__,__timestamp__:rfc3339&match[]={__name__=\"test_csv\"}&step=30s&start={TIME_MS-180s}"
+  ],
+  "expected_result_lines_count": 4
+}

app/victoria-metrics/testdata/csv/with_extra_labels.json

@@ -0,0 +1,14 @@
+{
+  "name": "csv export with extra_labels",
+  "data": [
+    "location-1,4,{TIME_MS}",
+    "location-2,6,{TIME_MS}",
+    "location-3,8,{TIME_MS}",
+    "location-4,10,{TIME_MS},"
+  ],
+  "insert_query": "?format=1:label:location,2:metric:test_csv_labels,3:time:unix_ms&extra_label=location=location-1",
+  "query": [
+    "/api/v1/export/csv?format=__name__,location,__value__,__timestamp__:unix_ms&match[]={__name__=\"test_csv\"}&step=30s&start={TIME_MS-180s}"
+  ],
+  "expected_result_lines_count": 4
+}

app/victoria-metrics/testdata/graphite/basic.json

@@ -0,0 +1,8 @@
+{
+  "name": "basic_insertion",
+  "data": ["graphite.foo.bar.baz;tag1=value1;tag2=value2 123 {TIME_S}"],
+  "query": ["/api/v1/export?match={__name__!=''}"],
+  "result_metrics": [
+    {"metric":{"__name__":"graphite.foo.bar.baz","tag1":"value1","tag2":"value2"},"values":[123], "timestamps": ["{TIME_MSZ}"]}
+  ]
+}

app/victoria-metrics/testdata/graphite/comparison-not-inf-not-nan.json

@@ -0,0 +1,16 @@
+{
+  "name": "comparison-not-inf-not-nan",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/150",
+  "data": [
+    "not_nan_not_inf;item=x 1 {TIME_S-1m}",
+    "not_nan_not_inf;item=x 1 {TIME_S-2m}",
+    "not_nan_not_inf;item=y 3 {TIME_S-1m}",
+    "not_nan_not_inf;item=y 1 {TIME_S-2m}"],
+  "query": ["/api/v1/query_range?query=1/(not_nan_not_inf-1)!=inf!=nan&start={TIME_S-3m}&end={TIME_S}&step=60"],
+  "result_query": {
+    "status":"success",
+    "data":{"resultType":"matrix",
+      "result":[
+	      {"metric":{"item":"y"},"values":[["{TIME_S-1m}","0.5"], ["{TIME_S}","0.5"]]}
+      ]}}
+}

app/victoria-metrics/testdata/graphite/empty-label-match.json

@@ -0,0 +1,16 @@
+{
+  "name": "empty-label-match",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/395",
+  "data": [
+    "empty_label_match 1 {TIME_S-1m}",
+    "empty_label_match;foo=bar 2 {TIME_S-1m}",
+    "empty_label_match;foo=baz 3 {TIME_S-1m}"],
+  "query": ["/api/v1/query_range?query=empty_label_match{foo=~'bar|'}&start={TIME_S-1m}&end={TIME_S}&step=60"],
+  "result_query": {
+    "status":"success",
+    "data":{"resultType":"matrix",
+      "result":[
+	      {"metric":{"__name__":"empty_label_match"},"values":[["{TIME_S-1m}","1"],["{TIME_S}","1"]]},
+	      {"metric":{"__name__":"empty_label_match","foo":"bar"},"values":[["{TIME_S-1m}","2"],["{TIME_S}","2"]]}
+      ]}}
+}

app/victoria-metrics/testdata/graphite/graphite-selector.json

@@ -0,0 +1,17 @@
+{
+  "name": "graphite-selector",
+  "issue": "",
+  "data": [
+    "graphite-selector.bar.baz 1 {TIME_S-1m}",
+    "graphite-selector.xxx.yy  2 {TIME_S-1m}",
+    "graphite-selector.bb.cc 3 {TIME_S-1m}",
+    "graphite-selector.a.baz 4 {TIME_S-1m}"],
+  "query": ["/api/v1/query?query=sort({__graphite__='graphite-selector.*.baz'})&time={TIME_S-1m}"],
+  "result_query": {
+    "status":"success",
+    "data":{"resultType":"vector","result":[
+	    {"metric":{"__name__":"graphite-selector.bar.baz"},"value":["{TIME_S-1m}","1"]},
+	    {"metric":{"__name__":"graphite-selector.a.baz"},"value":["{TIME_S-1m}","4"]}
+    ]}
+  }
+}

app/victoria-metrics/testdata/graphite/max_lookback_set.json

@@ -0,0 +1,23 @@
+{
+  "name": "max_lookback_set",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/209",
+  "data": [
+    "max_lookback_set 1 {TIME_S-30s}",
+    "max_lookback_set 2 {TIME_S-60s}",
+    "max_lookback_set 3 {TIME_S-120s}",
+    "max_lookback_set 4 {TIME_S-150s}"
+  ],
+  "query": ["/api/v1/query_range?query=max_lookback_set&start={TIME_S-150s}&end={TIME_S}&step=10s&max_lookback=1s"],
+  "result_query": {
+    "status":"success",
+    "data":{"resultType":"matrix",
+      "result":[{"metric":{"__name__":"max_lookback_set"},"values":[
+	      ["{TIME_S-150s}","4"],
+	      ["{TIME_S-120s}","3"],
+	      ["{TIME_S-60s}","2"],
+	      ["{TIME_S-30s}","1"],
+	      ["{TIME_S-20s}","1"],
+	      ["{TIME_S-10s}","1"],
+	      ["{TIME_S-0s}","1"]
+      ]}]}}
+}

app/victoria-metrics/testdata/graphite/max_lookback_unset.json

@@ -0,0 +1,31 @@
+{
+  "name": "max_lookback_unset",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/209",
+  "data": [
+    "max_lookback_unset 1 {TIME_S-30s}",
+    "max_lookback_unset 2 {TIME_S-60s}",
+    "max_lookback_unset 3 {TIME_S-120s}",
+    "max_lookback_unset 4 {TIME_S-150s}"
+  ],
+  "query": ["/api/v1/query_range?query=max_lookback_unset&start={TIME_S-150s}&end={TIME_S}&step=10s"],
+  "result_query": {
+    "status":"success",
+    "data":{"resultType":"matrix",
+      "result":[{"metric":{"__name__":"max_lookback_unset"},"values":[
+	      ["{TIME_S-150s}","4"],
+	      ["{TIME_S-140s}","4"],
+	      ["{TIME_S-130s}","4"],
+	      ["{TIME_S-120s}","3"],
+	      ["{TIME_S-110s}","3"],
+	      ["{TIME_S-100s}","3"],
+	      ["{TIME_S-90s}","3"],
+	      ["{TIME_S-80s}","3"],
+	      ["{TIME_S-60s}","2"],
+	      ["{TIME_S-50s}","2"],
+	      ["{TIME_S-40s}","2"],
+	      ["{TIME_S-30s}","1"],
+	      ["{TIME_S-20s}","1"],
+	      ["{TIME_S-10s}","1"],
+	      ["{TIME_S-0s}","1"]
+      ]}]}}
+}

app/victoria-metrics/testdata/graphite/name-plus-negative-filter.json

@@ -0,0 +1,16 @@
+{
+  "name": "name-plus-negative-filter",
+  "issue": "",
+  "data": [
+    "name-plus-negative-filter;foo=123 1 {TIME_S-1m}",
+    "name-plus-negative-filter;bar=123 2 {TIME_S-1m}",
+    "name-plus-negative-filter;foo=qwe 3 {TIME_S-1m}"
+  ],
+  "query": ["/api/v1/query?query={__name__='name-plus-negative-filter',foo!='123'}&time={TIME_S-1m}"],
+  "result_query": {
+    "status":"success",
+    "data":{"resultType":"vector","result":[
+	    {"metric":{"__name__":"name-plus-negative-filter","foo":"qwe"},"value":["{TIME_S-1m}","3"]}
+    ]}
+  }
+}

app/victoria-metrics/testdata/graphite/not-nan-as-missing-data.json

@@ -0,0 +1,18 @@
+{
+  "name": "not-nan-as-missing-data",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/153",
+  "data": [
+    "not_nan_as_missing_data;item=x 2 {TIME_S-2m}",
+    "not_nan_as_missing_data;item=x 1 {TIME_S-1m}",
+    "not_nan_as_missing_data;item=y 4 {TIME_S-2m}",
+    "not_nan_as_missing_data;item=y 3 {TIME_S-1m}"
+  ],
+  "query": ["/api/v1/query_range?query=not_nan_as_missing_data>1&start={TIME_S-2m}&end={TIME_S}&step=60"],
+  "result_query": {
+    "status":"success",
+    "data":{"resultType":"matrix",
+      "result":[
+	      {"metric":{"__name__":"not_nan_as_missing_data","item":"x"},"values":[["{TIME_S-2m}","2"]]},
+	      {"metric":{"__name__":"not_nan_as_missing_data","item":"y"},"values":[["{TIME_S-2m}","4"],["{TIME_S-1m}","3"],["{TIME_S}", "3"]]}
+      ]}}
+}

app/victoria-metrics/testdata/graphite/subquery-aggregation.json

@@ -0,0 +1,14 @@
+{
+  "name": "subquery-aggregation",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/184",
+  "data": [
+    "forms_daily_count;item=x 1 {TIME_S-1m}",
+    "forms_daily_count;item=x 2 {TIME_S-2m}",
+    "forms_daily_count;item=y 3 {TIME_S-1m}",
+    "forms_daily_count;item=y 4 {TIME_S-2m}"],
+  "query": ["/api/v1/query?query=min%20by%20(item)%20(min_over_time(forms_daily_count[10m:1m]))&time={TIME_S-1m}&latency_offset=1ms"],
+  "result_query": {
+    "status":"success",
+    "data":{"resultType":"vector","result":[{"metric":{"item":"x"},"value":["{TIME_S-1m}","2"]},{"metric":{"item":"y"},"value":["{TIME_S-1m}","4"]}]}
+  }
+}

app/victoria-metrics/testdata/influxdb/basic.json

@@ -0,0 +1,9 @@
+{
+  "name": "basic_insertion",
+  "data": ["measurement,tag1=value1,tag2=value2 field1=1.23,field2=123 {TIME_NS}"],
+  "query": ["/api/v1/export?match={__name__!=''}"],
+  "result_metrics": [
+    {"metric":{"__name__":"measurement_field2","tag1":"value1","tag2":"value2"},"values":[123], "timestamps": ["{TIME_MS}"]},
+    {"metric":{"__name__":"measurement_field1","tag1":"value1","tag2":"value2"},"values":[1.23], "timestamps": ["{TIME_MS}"]}
+  ]
+}

app/victoria-metrics/testdata/influxdb/with_extra_labels.json

@@ -0,0 +1,10 @@
+{
+  "name": "insert_with_extra_labels",
+  "data": ["measurement,tag1=value1,tag2=value2 field6=1.23,field5=123 {TIME_NS}"],
+  "insert_query": "?extra_label=job=test&extra_label=tag2=value10",
+  "query": ["/api/v1/export?match={__name__!=''}"],
+  "result_metrics": [
+    {"metric":{"__name__":"measurement_field5","tag1":"value1","job": "test","tag2":"value10"},"values":[123], "timestamps": ["{TIME_MS}"]},
+    {"metric":{"__name__":"measurement_field6","tag1":"value1","job": "test","tag2":"value10"},"values":[1.23], "timestamps": ["{TIME_MS}"]}
+  ]
+}

app/victoria-metrics/testdata/opentsdb/basic.json

@@ -0,0 +1,8 @@
+{
+  "name": "basic_insertion",
+  "data": ["put openstdb.foo.bar.baz {TIME_S} 123 tag1=value1 tag2=value2"],
+  "query": ["/api/v1/export?match={__name__!=''}"],
+  "result_metrics": [
+    {"metric":{"__name__":"openstdb.foo.bar.baz","tag1":"value1","tag2":"value2"},"values":[123], "timestamps": ["{TIME_MSZ}"]}
+  ]
+}

app/victoria-metrics/testdata/opentsdbhttp/basic.json

@@ -0,0 +1,8 @@
+{
+  "name": "basic_insertion",
+  "data": ["{\"metric\": \"opentsdbhttp.foo\", \"value\": 1001, \"timestamp\": {TIME_S}, \"tags\": {\"bar\":\"baz\", \"x\": \"y\"}}"],
+  "query": ["/api/v1/export?match={__name__!=''}"],
+  "result_metrics": [
+    {"metric":{"__name__":"opentsdbhttp.foo","bar":"baz","x":"y"},"values":[1001], "timestamps": ["{TIME_MSZ}"]}
+  ]
+}

app/victoria-metrics/testdata/opentsdbhttp/multi_line.json

@@ -0,0 +1,9 @@
+{
+  "name": "multiline",
+  "data": ["[{\"metric\": \"opentsdbhttp.multiline1\", \"value\": 1001, \"timestamp\": \"{TIME_S}\", \"tags\": {\"bar\":\"baz\", \"x\": \"y\"}}, {\"metric\": \"opentsdbhttp.multiline2\", \"value\": 1002, \"timestamp\": {TIME_S}}]"],
+  "query": ["/api/v1/export?match={__name__!=''}"],
+  "result_metrics": [
+    {"metric":{"__name__":"opentsdbhttp.multiline1","bar":"baz","x":"y"},"values":[1001], "timestamps": ["{TIME_MSZ}"]},
+    {"metric":{"__name__":"opentsdbhttp.multiline2"},"values":[1002], "timestamps": ["{TIME_MSZ}"]}
+  ]
+}

app/victoria-metrics/testdata/opentsdbhttp/with_extra_labels.json

@@ -0,0 +1,9 @@
+{
+  "name": "insert_with_extra_labels",
+  "data": ["{\"metric\": \"opentsdbhttp.foobar\", \"value\": 1001, \"timestamp\": {TIME_S}, \"tags\": {\"bar\":\"baz\", \"x\": \"y\"}}"],
+  "insert_query": "?extra_label=job=open-test&extra_label=x=z",
+  "query": ["/api/v1/export?match={__name__!=''}"],
+  "result_metrics": [
+    {"metric":{"__name__":"opentsdbhttp.foobar","bar":"baz","x":"z","job": "open-test"},"values":[1001], "timestamps": ["{TIME_MSZ}"]}
+  ]
+}

app/victoria-metrics/testdata/prometheus/basic.json

@@ -0,0 +1,8 @@
+{
+  "name": "basic_insertion",
+  "data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.bar\"},{\"name\":\"baz\",\"value\":\"qux\"}],\"samples\":[{\"value\":100000,\"timestamp\":\"{TIME_MS}\"}]}]"],
+  "query": ["/api/v1/export?match={__name__!=''}"],
+  "result_metrics": [
+    {"metric":{"__name__":"prometheus.bar","baz":"qux"},"values":[100000], "timestamps": ["{TIME_MS}"]}
+  ]
+}

app/victoria-metrics/testdata/prometheus/case-sensitive-regex.json

@@ -0,0 +1,10 @@
+{
+  "name": "case-sensitive-regex",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/161",
+  "data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.sensitiveRegex\"},{\"name\":\"label\",\"value\":\"sensitiveRegex\"}],\"samples\":[{\"value\":2,\"timestamp\":\"{TIME_MS}\"}]},{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.sensitiveRegex\"},{\"name\":\"label\",\"value\":\"SensitiveRegex\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]}]"],
+  "query": ["/api/v1/export?match={label=~'(?i)sensitiveregex'}"],
+  "result_metrics": [
+    {"metric":{"__name__":"prometheus.sensitiveRegex","label":"sensitiveRegex"},"values":[2], "timestamps": ["{TIME_MS}"]},
+    {"metric":{"__name__":"prometheus.sensitiveRegex","label":"SensitiveRegex"},"values":[1], "timestamps": ["{TIME_MS}"]}
+  ]
+}

app/victoria-metrics/testdata/prometheus/duplicate-label.json

@@ -0,0 +1,9 @@
+{
+  "name": "duplicate_label",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/172",
+  "data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.duplicate_label\"},{\"name\":\"duplicate\",\"value\":\"label\"},{\"name\":\"duplicate\",\"value\":\"label\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]}]"],
+  "query": ["/api/v1/export?match={__name__!=''}"],
+  "result_metrics": [
+    {"metric":{"__name__":"prometheus.duplicate_label","duplicate":"label"},"values":[1], "timestamps": ["{TIME_MS}"]}
+  ]
+}

app/victoria-metrics/testdata/prometheus/instant-matrix.json

@@ -0,0 +1,12 @@
+{
+  "name": "instant query with look-behind window",
+  "data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"foo\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS-60s}\"}]}]"],
+  "query": ["/api/v1/query?query=foo[5m]"],
+  "result_query": {
+      "status": "success",
+      "data":{
+        "resultType":"matrix",
+        "result":[{"metric":{"__name__":"foo"},"values":[["{TIME_S-60s}", "1"]]}]
+      }
+    }
+}

app/victoria-metrics/testdata/prometheus/instant-scalar.json

@@ -0,0 +1,11 @@
+{
+  "name": "instant scalar query",
+  "query": ["/api/v1/query?query=42&time={TIME_S}"],
+  "result_query": {
+      "status": "success",
+      "data":{
+        "resultType":"vector",
+        "result":[{"metric":{},"value":["{TIME_S}", "42"]}]
+      }
+    }
+}

app/victoria-metrics/testdata/prometheus/issue-5553-too-big-lookback.json

@@ -0,0 +1,13 @@
+{
+  "name": "too big look-behind window",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5553",
+  "data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"foo\"},{\"name\":\"issue\",\"value\":\"5553\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS-60s}\"}]}]"],
+  "query": ["/api/v1/query?query=foo{issue=\"5553\"}[100y]"],
+  "result_query": {
+      "status": "success",
+      "data":{
+        "resultType":"matrix",
+        "result":[{"metric":{"__name__":"foo", "issue": "5553"},"values":[["{TIME_S-60s}", "1"]]}]
+      }
+    }
+}

app/victoria-metrics/testdata/prometheus/match-series.json

@@ -0,0 +1,15 @@
+{
+  "name": "match_series",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/155",
+  "data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"MatchSeries\"},{\"name\":\"db\",\"value\":\"TenMinute\"},{\"name\":\"TurbineType\",\"value\":\"V112\"},{\"name\":\"Park\",\"value\":\"1\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]},{\"labels\":[{\"name\":\"__name__\",\"value\":\"MatchSeries\"},{\"name\":\"db\",\"value\":\"TenMinute\"},{\"name\":\"TurbineType\",\"value\":\"V112\"},{\"name\":\"Park\",\"value\":\"2\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]},{\"labels\":[{\"name\":\"__name__\",\"value\":\"MatchSeries\"},{\"name\":\"db\",\"value\":\"TenMinute\"},{\"name\":\"TurbineType\",\"value\":\"V112\"},{\"name\":\"Park\",\"value\":\"3\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]},{\"labels\":[{\"name\":\"__name__\",\"value\":\"MatchSeries\"},{\"name\":\"db\",\"value\":\"TenMinute\"},{\"name\":\"TurbineType\",\"value\":\"V112\"},{\"name\":\"Park\",\"value\":\"4\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]}]"],
+  "query": ["/api/v1/series?match[]={__name__='MatchSeries'}", "/api/v1/series?match[]={__name__=~'MatchSeries.*'}"],
+  "result_series": {
+    "status": "success",
+    "data": [
+      {"__name__":"MatchSeries","db":"TenMinute","Park":"1","TurbineType":"V112"},
+      {"__name__":"MatchSeries","db":"TenMinute","Park":"2","TurbineType":"V112"},
+      {"__name__":"MatchSeries","db":"TenMinute","Park":"3","TurbineType":"V112"},
+      {"__name__":"MatchSeries","db":"TenMinute","Park":"4","TurbineType":"V112"}
+    ]
+  }
+}

app/victoria-metrics/testdata/prometheus/query-range.json

@@ -0,0 +1,18 @@
+{
+  "name": "query range",
+  "issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5553",
+  "data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"bar\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS-60s}\"}, {\"value\":2,\"timestamp\":\"{TIME_MS-120s}\"}, {\"value\":1,\"timestamp\":\"{TIME_MS-180s}\"}]}]"],
+  "query": ["/api/v1/query_range?query=bar&step=30s&start={TIME_MS-180s}"],
+  "result_query": {
+      "status": "success",
+      "data":{
+        "resultType":"matrix",
+        "result":[
+          {
+            "metric":{"__name__":"bar"},
+            "values":[["{TIME_S-180s}", "1"],["{TIME_S-150s}", "1"],["{TIME_S-120s}", "2"],["{TIME_S-90s}", "2"], ["{TIME_S-60s}", "1"], ["{TIME_S-30s}", "1"], ["{TIME_S}", "1"]]
+          }
+        ]
+      }
+    }
+}

app/victoria-metrics/testdata/prometheus/with_extra_labels.json

@@ -0,0 +1,9 @@
+{
+  "name": "basic_insertion_with_extra_labels",
+  "insert_query": "?extra_label=job=prom-test&extra_label=baz=bar",
+  "data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.foobar\"},{\"name\":\"baz\",\"value\":\"qux\"}],\"samples\":[{\"value\":100000,\"timestamp\":\"{TIME_MS}\"}]}]"],
+  "query": ["/api/v1/export?match={__name__!=''}"],
+  "result_metrics": [
+    {"metric":{"__name__":"prometheus.foobar","baz":"bar","job": "prom-test"},"values":[100000], "timestamps": ["{TIME_MS}"]}
+  ]
+}

app/victoria-metrics/testdata/prometheus/with_request_extra_filter.json

@@ -0,0 +1,8 @@
+{
+  "name": "basic_select_with_extra_labels",
+  "data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.tenant.limits\"},{\"name\":\"baz\",\"value\":\"qux\"},{\"name\":\"tenant\",\"value\":\"dev\"}],\"samples\":[{\"value\":100000,\"timestamp\":\"{TIME_MS}\"}]},{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.up\"},{\"name\":\"baz\",\"value\":\"qux\"}],\"samples\":[{\"value\":100000,\"timestamp\":\"{TIME_MS}\"}]}]"],
+  "query": ["/api/v1/export?match={__name__!=''}&extra_label=tenant=dev"],
+  "result_metrics": [
+    {"metric":{"__name__":"prometheus.tenant.limits","baz":"qux","tenant": "dev"},"values":[100000], "timestamps": ["{TIME_MS}"]}
+  ]
+}

app/vlagent/README.md

@@ -1 +0,0 @@
-VictoriaLogs source code has been moved to [github.com/VictoriaMetrics/VictoriaLogs](https://github.com/VictoriaMetrics/VictoriaLogs/).

app/vlinsert/README.md

@@ -1 +0,0 @@
-VictoriaLogs source code has been moved to [github.com/VictoriaMetrics/VictoriaLogs](https://github.com/VictoriaMetrics/VictoriaLogs/).

app/vlinsert/elasticsearch/bulk_response.qtpl

@@ -0,0 +1,20 @@
+{% stripspace %}
+
+{% func BulkResponse(n int, tookMs int64) %}
+{
+	"took":{%dl tookMs %},
+	"errors":false,
+	"items":[
+		{% for i := 0; i < n; i++ %}
+		{
+			"create":{
+				"status":201
+			}
+		}
+		{% if i+1 < n %},{% endif %}
+		{% endfor %}
+	]
+}
+{% endfunc %}
+
+{% endstripspace %}

app/vlinsert/elasticsearch/bulk_response.qtpl.go

@@ -0,0 +1,69 @@
+// Code generated by qtc from "bulk_response.qtpl". DO NOT EDIT.
+// See https://github.com/valyala/quicktemplate for details.
+
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:3
+package elasticsearch
+
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:3
+import (
+	qtio422016 "io"
+
+	qt422016 "github.com/valyala/quicktemplate"
+)
+
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:3
+var (
+	_ = qtio422016.Copy
+	_ = qt422016.AcquireByteBuffer
+)
+
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:3
+func StreamBulkResponse(qw422016 *qt422016.Writer, n int, tookMs int64) {
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:3
+	qw422016.N().S(`{"took":`)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:5
+	qw422016.N().DL(tookMs)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:5
+	qw422016.N().S(`,"errors":false,"items":[`)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:8
+	for i := 0; i < n; i++ {
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:8
+		qw422016.N().S(`{"create":{"status":201}}`)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:14
+		if i+1 < n {
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:14
+			qw422016.N().S(`,`)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:14
+		}
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:15
+	}
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:15
+	qw422016.N().S(`]}`)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+}
+
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+func WriteBulkResponse(qq422016 qtio422016.Writer, n int, tookMs int64) {
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+	qw422016 := qt422016.AcquireWriter(qq422016)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+	StreamBulkResponse(qw422016, n, tookMs)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+	qt422016.ReleaseWriter(qw422016)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+}
+
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+func BulkResponse(n int, tookMs int64) string {
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+	qb422016 := qt422016.AcquireByteBuffer()
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+	WriteBulkResponse(qb422016, n, tookMs)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+	qs422016 := string(qb422016.B)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+	qt422016.ReleaseByteBuffer(qb422016)
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+	return qs422016
+//line app/vlinsert/elasticsearch/bulk_response.qtpl:18
+}

app/vlinsert/elasticsearch/elasticsearch.go

@@ -0,0 +1,281 @@
+package elasticsearch
+
+import (
+	"bufio"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"math"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/VictoriaMetrics/metrics"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bufferedwriter"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logjson"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
+)
+
+var (
+	elasticsearchVersion = flag.String("elasticsearch.version", "8.9.0", "Elasticsearch version to report to client")
+)
+
+// RequestHandler processes Elasticsearch insert requests
+func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
+	w.Header().Add("Content-Type", "application/json")
+	// This header is needed for Logstash
+	w.Header().Set("X-Elastic-Product", "Elasticsearch")
+
+	if strings.HasPrefix(path, "/_ilm/policy") {
+		// Return fake response for Elasticsearch ilm request.
+		fmt.Fprintf(w, `{}`)
+		return true
+	}
+	if strings.HasPrefix(path, "/_index_template") {
+		// Return fake response for Elasticsearch index template request.
+		fmt.Fprintf(w, `{}`)
+		return true
+	}
+	if strings.HasPrefix(path, "/_ingest") {
+		// Return fake response for Elasticsearch ingest pipeline request.
+		// See: https://www.elastic.co/guide/en/elasticsearch/reference/8.8/put-pipeline-api.html
+		fmt.Fprintf(w, `{}`)
+		return true
+	}
+	if strings.HasPrefix(path, "/_nodes") {
+		// Return fake response for Elasticsearch nodes discovery request.
+		// See: https://www.elastic.co/guide/en/elasticsearch/reference/8.8/cluster.html
+		fmt.Fprintf(w, `{}`)
+		return true
+	}
+	switch path {
+	case "/":
+		switch r.Method {
+		case http.MethodGet:
+			// Return fake response for Elasticsearch ping request.
+			// See the latest available version for Elasticsearch at https://github.com/elastic/elasticsearch/releases
+			fmt.Fprintf(w, `{
+			"version": {
+				"number": %q
+			}
+		}`, *elasticsearchVersion)
+		case http.MethodHead:
+			// Return empty response for Logstash ping request.
+		}
+
+		return true
+	case "/_license":
+		// Return fake response for Elasticsearch license request.
+		fmt.Fprintf(w, `{
+			"license": {
+				"uid": "cbff45e7-c553-41f7-ae4f-9205eabd80xx",
+				"type": "oss",
+				"status": "active",
+				"expiry_date_in_millis" : 4000000000000
+			}
+		}`)
+		return true
+	case "/_bulk":
+		startTime := time.Now()
+		bulkRequestsTotal.Inc()
+
+		cp, err := insertutils.GetCommonParams(r)
+		if err != nil {
+			httpserver.Errorf(w, r, "%s", err)
+			return true
+		}
+		if err := vlstorage.CanWriteData(); err != nil {
+			httpserver.Errorf(w, r, "%s", err)
+			return true
+		}
+		lr := logstorage.GetLogRows(cp.StreamFields, cp.IgnoreFields)
+		processLogMessage := cp.GetProcessLogMessageFunc(lr)
+		isGzip := r.Header.Get("Content-Encoding") == "gzip"
+		n, err := readBulkRequest(r.Body, isGzip, cp.TimeField, cp.MsgField, processLogMessage)
+		vlstorage.MustAddRows(lr)
+		logstorage.PutLogRows(lr)
+		if err != nil {
+			logger.Warnf("cannot decode log message #%d in /_bulk request: %s, stream fields: %s", n, err, cp.StreamFields)
+			return true
+		}
+
+		tookMs := time.Since(startTime).Milliseconds()
+		bw := bufferedwriter.Get(w)
+		defer bufferedwriter.Put(bw)
+		WriteBulkResponse(bw, n, tookMs)
+		_ = bw.Flush()
+
+		// update bulkRequestDuration only for successfully parsed requests
+		// There is no need in updating bulkRequestDuration for request errors,
+		// since their timings are usually much smaller than the timing for successful request parsing.
+		bulkRequestDuration.UpdateDuration(startTime)
+
+		return true
+	default:
+		return false
+	}
+}
+
+var (
+	bulkRequestsTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/elasticsearch/_bulk"}`)
+	rowsIngestedTotal   = metrics.NewCounter(`vl_rows_ingested_total{type="elasticsearch_bulk"}`)
+	bulkRequestDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/elasticsearch/_bulk"}`)
+)
+
+func readBulkRequest(r io.Reader, isGzip bool, timeField, msgField string,
+	processLogMessage func(timestamp int64, fields []logstorage.Field),
+) (int, error) {
+	// See https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html
+
+	if isGzip {
+		zr, err := common.GetGzipReader(r)
+		if err != nil {
+			return 0, fmt.Errorf("cannot read gzipped _bulk request: %w", err)
+		}
+		defer common.PutGzipReader(zr)
+		r = zr
+	}
+
+	wcr := writeconcurrencylimiter.GetReader(r)
+	defer writeconcurrencylimiter.PutReader(wcr)
+
+	lb := lineBufferPool.Get()
+	defer lineBufferPool.Put(lb)
+
+	lb.B = bytesutil.ResizeNoCopyNoOverallocate(lb.B, insertutils.MaxLineSizeBytes.IntN())
+	sc := bufio.NewScanner(wcr)
+	sc.Buffer(lb.B, len(lb.B))
+
+	n := 0
+	nCheckpoint := 0
+	for {
+		ok, err := readBulkLine(sc, timeField, msgField, processLogMessage)
+		wcr.DecConcurrency()
+		if err != nil || !ok {
+			rowsIngestedTotal.Add(n - nCheckpoint)
+			return n, err
+		}
+		n++
+		if batchSize := n - nCheckpoint; n >= 1000 {
+			rowsIngestedTotal.Add(batchSize)
+			nCheckpoint = n
+		}
+	}
+}
+
+var lineBufferPool bytesutil.ByteBufferPool
+
+func readBulkLine(sc *bufio.Scanner, timeField, msgField string,
+	processLogMessage func(timestamp int64, fields []logstorage.Field),
+) (bool, error) {
+	var line []byte
+
+	// Read the command, must be "create" or "index"
+	for len(line) == 0 {
+		if !sc.Scan() {
+			if err := sc.Err(); err != nil {
+				if errors.Is(err, bufio.ErrTooLong) {
+					return false, fmt.Errorf(`cannot read "create" or "index" command, since its size exceeds -insert.maxLineSizeBytes=%d`,
+						insertutils.MaxLineSizeBytes.IntN())
+				}
+				return false, err
+			}
+			return false, nil
+		}
+		line = sc.Bytes()
+	}
+	lineStr := bytesutil.ToUnsafeString(line)
+	if !strings.Contains(lineStr, `"create"`) && !strings.Contains(lineStr, `"index"`) {
+		return false, fmt.Errorf(`unexpected command %q; expecting "create" or "index"`, line)
+	}
+
+	// Decode log message
+	if !sc.Scan() {
+		if err := sc.Err(); err != nil {
+			if errors.Is(err, bufio.ErrTooLong) {
+				return false, fmt.Errorf("cannot read log message, since its size exceeds -insert.maxLineSizeBytes=%d", insertutils.MaxLineSizeBytes.IntN())
+			}
+			return false, err
+		}
+		return false, fmt.Errorf(`missing log message after the "create" or "index" command`)
+	}
+	line = sc.Bytes()
+	p := logjson.GetParser()
+	if err := p.ParseLogMessage(line); err != nil {
+		return false, fmt.Errorf("cannot parse json-encoded log entry: %w", err)
+	}
+
+	ts, err := extractTimestampFromFields(timeField, p.Fields)
+	if err != nil {
+		return false, fmt.Errorf("cannot parse timestamp: %w", err)
+	}
+	if ts == 0 {
+		ts = time.Now().UnixNano()
+	}
+	p.RenameField(msgField, "_msg")
+	processLogMessage(ts, p.Fields)
+	logjson.PutParser(p)
+
+	return true, nil
+}
+
+func extractTimestampFromFields(timeField string, fields []logstorage.Field) (int64, error) {
+	for i := range fields {
+		f := &fields[i]
+		if f.Name != timeField {
+			continue
+		}
+		timestamp, err := parseElasticsearchTimestamp(f.Value)
+		if err != nil {
+			return 0, err
+		}
+		f.Value = ""
+		return timestamp, nil
+	}
+	return 0, nil
+}
+
+func parseElasticsearchTimestamp(s string) (int64, error) {
+	if s == "0" || s == "" {
+		// Special case - zero or empty timestamp must be substituted
+		// with the current time by the caller.
+		return 0, nil
+	}
+	if len(s) < len("YYYY-MM-DD") || s[len("YYYY")] != '-' {
+		// Try parsing timestamp in milliseconds
+		n, err := strconv.ParseInt(s, 10, 64)
+		if err != nil {
+			return 0, fmt.Errorf("cannot parse timestamp in milliseconds from %q: %w", s, err)
+		}
+		if n > int64(math.MaxInt64)/1e6 {
+			return 0, fmt.Errorf("too big timestamp in milliseconds: %d; mustn't exceed %d", n, int64(math.MaxInt64)/1e6)
+		}
+		if n < int64(math.MinInt64)/1e6 {
+			return 0, fmt.Errorf("too small timestamp in milliseconds: %d; must be bigger than %d", n, int64(math.MinInt64)/1e6)
+		}
+		n *= 1e6
+		return n, nil
+	}
+	if len(s) == len("YYYY-MM-DD") {
+		t, err := time.Parse("2006-01-02", s)
+		if err != nil {
+			return 0, fmt.Errorf("cannot parse date %q: %w", s, err)
+		}
+		return t.UnixNano(), nil
+	}
+	t, err := time.Parse(time.RFC3339, s)
+	if err != nil {
+		return 0, fmt.Errorf("cannot parse timestamp %q: %w", s, err)
+	}
+	return t.UnixNano(), nil
+}

app/vlinsert/elasticsearch/elasticsearch_test.go

@@ -0,0 +1,129 @@
+package elasticsearch
+
+import (
+	"bytes"
+	"compress/gzip"
+	"fmt"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+)
+
+func TestReadBulkRequestFailure(t *testing.T) {
+	f := func(data string) {
+		t.Helper()
+
+		processLogMessage := func(timestamp int64, fields []logstorage.Field) {
+			t.Fatalf("unexpected call to processLogMessage with timestamp=%d, fields=%s", timestamp, fields)
+		}
+
+		r := bytes.NewBufferString(data)
+		rows, err := readBulkRequest(r, false, "_time", "_msg", processLogMessage)
+		if err == nil {
+			t.Fatalf("expecting non-empty error")
+		}
+		if rows != 0 {
+			t.Fatalf("unexpected non-zero rows=%d", rows)
+		}
+	}
+	f("foobar")
+	f(`{}`)
+	f(`{"create":{}}`)
+	f(`{"creat":{}}
+{}`)
+	f(`{"create":{}}
+foobar`)
+}
+
+func TestReadBulkRequestSuccess(t *testing.T) {
+	f := func(data, timeField, msgField string, rowsExpected int, timestampsExpected []int64, resultExpected string) {
+		t.Helper()
+
+		var timestamps []int64
+		var result string
+		processLogMessage := func(timestamp int64, fields []logstorage.Field) {
+			timestamps = append(timestamps, timestamp)
+
+			a := make([]string, len(fields))
+			for i, f := range fields {
+				a[i] = fmt.Sprintf("%q:%q", f.Name, f.Value)
+			}
+			s := "{" + strings.Join(a, ",") + "}\n"
+			result += s
+		}
+
+		// Read the request without compression
+		r := bytes.NewBufferString(data)
+		rows, err := readBulkRequest(r, false, timeField, msgField, processLogMessage)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if rows != rowsExpected {
+			t.Fatalf("unexpected rows read; got %d; want %d", rows, rowsExpected)
+		}
+
+		if !reflect.DeepEqual(timestamps, timestampsExpected) {
+			t.Fatalf("unexpected timestamps;\ngot\n%d\nwant\n%d", timestamps, timestampsExpected)
+		}
+		if result != resultExpected {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, resultExpected)
+		}
+
+		// Read the request with compression
+		timestamps = nil
+		result = ""
+		compressedData := compressData(data)
+		r = bytes.NewBufferString(compressedData)
+		rows, err = readBulkRequest(r, true, timeField, msgField, processLogMessage)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if rows != rowsExpected {
+			t.Fatalf("unexpected rows read; got %d; want %d", rows, rowsExpected)
+		}
+
+		if !reflect.DeepEqual(timestamps, timestampsExpected) {
+			t.Fatalf("unexpected timestamps;\ngot\n%d\nwant\n%d", timestamps, timestampsExpected)
+		}
+		if result != resultExpected {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, resultExpected)
+		}
+	}
+
+	// Verify an empty data
+	f("", "_time", "_msg", 0, nil, "")
+	f("\n", "_time", "_msg", 0, nil, "")
+	f("\n\n", "_time", "_msg", 0, nil, "")
+
+	// Verify non-empty data
+	data := `{"create":{"_index":"filebeat-8.8.0"}}
+{"@timestamp":"2023-06-06T04:48:11.735Z","log":{"offset":71770,"file":{"path":"/var/log/auth.log"}},"message":"foobar"}
+{"create":{"_index":"filebeat-8.8.0"}}
+{"@timestamp":"2023-06-06T04:48:12.735Z","message":"baz"}
+{"index":{"_index":"filebeat-8.8.0"}}
+{"message":"xyz","@timestamp":"2023-06-06T04:48:13.735Z","x":"y"}
+`
+	timeField := "@timestamp"
+	msgField := "message"
+	rowsExpected := 3
+	timestampsExpected := []int64{1686026891735000000, 1686026892735000000, 1686026893735000000}
+	resultExpected := `{"@timestamp":"","log.offset":"71770","log.file.path":"/var/log/auth.log","_msg":"foobar"}
+{"@timestamp":"","_msg":"baz"}
+{"_msg":"xyz","@timestamp":"","x":"y"}
+`
+	f(data, timeField, msgField, rowsExpected, timestampsExpected, resultExpected)
+}
+
+func compressData(s string) string {
+	var bb bytes.Buffer
+	zw := gzip.NewWriter(&bb)
+	if _, err := zw.Write([]byte(s)); err != nil {
+		panic(fmt.Errorf("unexpected error when compressing data: %w", err))
+	}
+	if err := zw.Close(); err != nil {
+		panic(fmt.Errorf("unexpected error when closing gzip writer: %w", err))
+	}
+	return bb.String()
+}

app/vlinsert/elasticsearch/elasticsearch_timing_test.go

@@ -0,0 +1,50 @@
+package elasticsearch
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+)
+
+func BenchmarkReadBulkRequest(b *testing.B) {
+	b.Run("gzip:off", func(b *testing.B) {
+		benchmarkReadBulkRequest(b, false)
+	})
+	b.Run("gzip:on", func(b *testing.B) {
+		benchmarkReadBulkRequest(b, true)
+	})
+}
+
+func benchmarkReadBulkRequest(b *testing.B, isGzip bool) {
+	data := `{"create":{"_index":"filebeat-8.8.0"}}
+{"@timestamp":"2023-06-06T04:48:11.735Z","log":{"offset":71770,"file":{"path":"/var/log/auth.log"}},"message":"foobar"}
+{"create":{"_index":"filebeat-8.8.0"}}
+{"@timestamp":"2023-06-06T04:48:12.735Z","message":"baz"}
+{"create":{"_index":"filebeat-8.8.0"}}
+{"message":"xyz","@timestamp":"2023-06-06T04:48:13.735Z","x":"y"}
+`
+	if isGzip {
+		data = compressData(data)
+	}
+	dataBytes := bytesutil.ToUnsafeBytes(data)
+
+	timeField := "@timestamp"
+	msgField := "message"
+	processLogMessage := func(_ int64, _ []logstorage.Field) {}
+
+	b.ReportAllocs()
+	b.SetBytes(int64(len(data)))
+	b.RunParallel(func(pb *testing.PB) {
+		r := &bytes.Reader{}
+		for pb.Next() {
+			r.Reset(dataBytes)
+			_, err := readBulkRequest(r, isGzip, timeField, msgField, processLogMessage)
+			if err != nil {
+				panic(fmt.Errorf("unexpected error: %w", err))
+			}
+		}
+	})
+}

app/vlinsert/insertutils/common_params.go

@@ -0,0 +1,100 @@
+package insertutils
+
+import (
+	"net/http"
+
+	"github.com/VictoriaMetrics/metrics"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+)
+
+// CommonParams contains common HTTP parameters used by log ingestion APIs.
+//
+// See https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/#http-parameters
+type CommonParams struct {
+	TenantID     logstorage.TenantID
+	TimeField    string
+	MsgField     string
+	StreamFields []string
+	IgnoreFields []string
+
+	Debug           bool
+	DebugRequestURI string
+	DebugRemoteAddr string
+}
+
+// GetCommonParams returns CommonParams from r.
+func GetCommonParams(r *http.Request) (*CommonParams, error) {
+	// Extract tenantID
+	tenantID, err := logstorage.GetTenantIDFromRequest(r)
+	if err != nil {
+		return nil, err
+	}
+
+	// Extract time field name from _time_field query arg
+	var timeField = "_time"
+	if tf := r.FormValue("_time_field"); tf != "" {
+		timeField = tf
+	}
+
+	// Extract message field name from _msg_field query arg
+	var msgField = ""
+	if msgf := r.FormValue("_msg_field"); msgf != "" {
+		msgField = msgf
+	}
+
+	streamFields := httputils.GetArray(r, "_stream_fields")
+	ignoreFields := httputils.GetArray(r, "ignore_fields")
+
+	debug := httputils.GetBool(r, "debug")
+	debugRequestURI := ""
+	debugRemoteAddr := ""
+	if debug {
+		debugRequestURI = httpserver.GetRequestURI(r)
+		debugRemoteAddr = httpserver.GetQuotedRemoteAddr(r)
+	}
+
+	cp := &CommonParams{
+		TenantID:        tenantID,
+		TimeField:       timeField,
+		MsgField:        msgField,
+		StreamFields:    streamFields,
+		IgnoreFields:    ignoreFields,
+		Debug:           debug,
+		DebugRequestURI: debugRequestURI,
+		DebugRemoteAddr: debugRemoteAddr,
+	}
+	return cp, nil
+}
+
+// GetProcessLogMessageFunc returns a function, which adds parsed log messages to lr.
+func (cp *CommonParams) GetProcessLogMessageFunc(lr *logstorage.LogRows) func(timestamp int64, fields []logstorage.Field) {
+	return func(timestamp int64, fields []logstorage.Field) {
+		if len(fields) > *MaxFieldsPerLine {
+			rf := logstorage.RowFormatter(fields)
+			logger.Warnf("dropping log line with %d fields; it exceeds -insert.maxFieldsPerLine=%d; %s", len(fields), *MaxFieldsPerLine, rf)
+			rowsDroppedTotalTooManyFields.Inc()
+			return
+		}
+
+		lr.MustAdd(cp.TenantID, timestamp, fields)
+		if cp.Debug {
+			s := lr.GetRowString(0)
+			lr.ResetKeepSettings()
+			logger.Infof("remoteAddr=%s; requestURI=%s; ignoring log entry because of `debug` query arg: %s", cp.DebugRemoteAddr, cp.DebugRequestURI, s)
+			rowsDroppedTotalDebug.Inc()
+			return
+		}
+		if lr.NeedFlush() {
+			vlstorage.MustAddRows(lr)
+			lr.ResetKeepSettings()
+		}
+	}
+}
+
+var rowsDroppedTotalDebug = metrics.NewCounter(`vl_rows_dropped_total{reason="debug"}`)
+var rowsDroppedTotalTooManyFields = metrics.NewCounter(`vl_rows_dropped_total{reason="too_many_fields"}`)

app/vlinsert/insertutils/flags.go

@@ -0,0 +1,15 @@
+package insertutils
+
+import (
+	"flag"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
+)
+
+var (
+	// MaxLineSizeBytes is the maximum length of a single line for /insert/* handlers
+	MaxLineSizeBytes = flagutil.NewBytes("insert.maxLineSizeBytes", 256*1024, "The maximum size of a single line, which can be read by /insert/* handlers")
+
+	// MaxFieldsPerLine is the maximum number of fields per line for /insert/* handlers
+	MaxFieldsPerLine = flag.Int("insert.maxFieldsPerLine", 1000, "The maximum number of log fields per line, which can be read by /insert/* handlers")
+)

app/vlinsert/jsonline/jsonline.go

@@ -0,0 +1,161 @@
+package jsonline
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logjson"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+// RequestHandler processes jsonline insert requests
+func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
+	startTime := time.Now()
+	w.Header().Add("Content-Type", "application/json")
+
+	if r.Method != "POST" {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		return true
+	}
+
+	requestsTotal.Inc()
+
+	cp, err := insertutils.GetCommonParams(r)
+	if err != nil {
+		httpserver.Errorf(w, r, "%s", err)
+		return true
+	}
+	if err := vlstorage.CanWriteData(); err != nil {
+		httpserver.Errorf(w, r, "%s", err)
+		return true
+	}
+	lr := logstorage.GetLogRows(cp.StreamFields, cp.IgnoreFields)
+	processLogMessage := cp.GetProcessLogMessageFunc(lr)
+
+	reader := r.Body
+	if r.Header.Get("Content-Encoding") == "gzip" {
+		zr, err := common.GetGzipReader(reader)
+		if err != nil {
+			logger.Errorf("cannot read gzipped _bulk request: %s", err)
+			return true
+		}
+		defer common.PutGzipReader(zr)
+		reader = zr
+	}
+
+	wcr := writeconcurrencylimiter.GetReader(reader)
+	defer writeconcurrencylimiter.PutReader(wcr)
+
+	lb := lineBufferPool.Get()
+	defer lineBufferPool.Put(lb)
+
+	lb.B = bytesutil.ResizeNoCopyNoOverallocate(lb.B, insertutils.MaxLineSizeBytes.IntN())
+	sc := bufio.NewScanner(wcr)
+	sc.Buffer(lb.B, len(lb.B))
+
+	n := 0
+	for {
+		ok, err := readLine(sc, cp.TimeField, cp.MsgField, processLogMessage)
+		wcr.DecConcurrency()
+		if err != nil {
+			logger.Errorf("cannot read line #%d in /jsonline request: %s", n, err)
+			break
+		}
+		if !ok {
+			break
+		}
+		n++
+		rowsIngestedTotal.Inc()
+	}
+
+	vlstorage.MustAddRows(lr)
+	logstorage.PutLogRows(lr)
+
+	// update jsonlineRequestDuration only for successfully parsed requests.
+	// There is no need in updating jsonlineRequestDuration for request errors,
+	// since their timings are usually much smaller than the timing for successful request parsing.
+	jsonlineRequestDuration.UpdateDuration(startTime)
+
+	return true
+}
+
+func readLine(sc *bufio.Scanner, timeField, msgField string, processLogMessage func(timestamp int64, fields []logstorage.Field)) (bool, error) {
+	var line []byte
+	for len(line) == 0 {
+		if !sc.Scan() {
+			if err := sc.Err(); err != nil {
+				if errors.Is(err, bufio.ErrTooLong) {
+					return false, fmt.Errorf(`cannot read json line, since its size exceeds -insert.maxLineSizeBytes=%d`, insertutils.MaxLineSizeBytes.IntN())
+				}
+				return false, err
+			}
+			return false, nil
+		}
+		line = sc.Bytes()
+	}
+
+	p := logjson.GetParser()
+	if err := p.ParseLogMessage(line); err != nil {
+		return false, fmt.Errorf("cannot parse json-encoded log entry: %w", err)
+	}
+	ts, err := extractTimestampFromFields(timeField, p.Fields)
+	if err != nil {
+		return false, fmt.Errorf("cannot parse timestamp: %w", err)
+	}
+	if ts == 0 {
+		ts = time.Now().UnixNano()
+	}
+	p.RenameField(msgField, "_msg")
+	processLogMessage(ts, p.Fields)
+	logjson.PutParser(p)
+
+	return true, nil
+}
+
+func extractTimestampFromFields(timeField string, fields []logstorage.Field) (int64, error) {
+	for i := range fields {
+		f := &fields[i]
+		if f.Name != timeField {
+			continue
+		}
+		timestamp, err := parseISO8601Timestamp(f.Value)
+		if err != nil {
+			return 0, err
+		}
+		f.Value = ""
+		return timestamp, nil
+	}
+	return 0, nil
+}
+
+func parseISO8601Timestamp(s string) (int64, error) {
+	if s == "0" || s == "" {
+		// Special case for returning the current timestamp.
+		// It must be automatically converted to the current timestamp by the caller.
+		return 0, nil
+	}
+	t, err := time.Parse(time.RFC3339, s)
+	if err != nil {
+		return 0, fmt.Errorf("cannot parse timestamp %q: %w", s, err)
+	}
+	return t.UnixNano(), nil
+}
+
+var lineBufferPool bytesutil.ByteBufferPool
+
+var (
+	requestsTotal           = metrics.NewCounter(`vl_http_requests_total{path="/insert/jsonline"}`)
+	rowsIngestedTotal       = metrics.NewCounter(`vl_rows_ingested_total{type="jsonline"}`)
+	jsonlineRequestDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/jsonline"}`)
+)

app/vlinsert/jsonline/jsonline_test.go

@@ -0,0 +1,70 @@
+package jsonline
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func TestReadBulkRequestSuccess(t *testing.T) {
+	f := func(data, timeField, msgField string, rowsExpected int, timestampsExpected []int64, resultExpected string) {
+		t.Helper()
+
+		var timestamps []int64
+		var result string
+		processLogMessage := func(timestamp int64, fields []logstorage.Field) {
+			timestamps = append(timestamps, timestamp)
+
+			a := make([]string, len(fields))
+			for i, f := range fields {
+				a[i] = fmt.Sprintf("%q:%q", f.Name, f.Value)
+			}
+			s := "{" + strings.Join(a, ",") + "}\n"
+			result += s
+		}
+
+		// Read the request without compression
+		r := bytes.NewBufferString(data)
+		sc := bufio.NewScanner(r)
+		rows := 0
+		for {
+			ok, err := readLine(sc, timeField, msgField, processLogMessage)
+			if err != nil {
+				t.Fatalf("unexpected error: %s", err)
+			}
+			if !ok {
+				break
+			}
+			rows++
+		}
+		if rows != rowsExpected {
+			t.Fatalf("unexpected rows read; got %d; want %d", rows, rowsExpected)
+		}
+
+		if !reflect.DeepEqual(timestamps, timestampsExpected) {
+			t.Fatalf("unexpected timestamps;\ngot\n%d\nwant\n%d", timestamps, timestampsExpected)
+		}
+		if result != resultExpected {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, resultExpected)
+		}
+	}
+
+	// Verify non-empty data
+	data := `{"@timestamp":"2023-06-06T04:48:11.735Z","log":{"offset":71770,"file":{"path":"/var/log/auth.log"}},"message":"foobar"}
+{"@timestamp":"2023-06-06T04:48:12.735Z","message":"baz"}
+{"message":"xyz","@timestamp":"2023-06-06T04:48:13.735Z","x":"y"}
+`
+	timeField := "@timestamp"
+	msgField := "message"
+	rowsExpected := 3
+	timestampsExpected := []int64{1686026891735000000, 1686026892735000000, 1686026893735000000}
+	resultExpected := `{"@timestamp":"","log.offset":"71770","log.file.path":"/var/log/auth.log","_msg":"foobar"}
+{"@timestamp":"","_msg":"baz"}
+{"_msg":"xyz","@timestamp":"","x":"y"}
+`
+	f(data, timeField, msgField, rowsExpected, timestampsExpected, resultExpected)
+}

app/vlinsert/loki/loki.go

@@ -0,0 +1,58 @@
+package loki
+
+import (
+	"net/http"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+)
+
+// RequestHandler processes Loki insert requests
+func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
+	switch path {
+	case "/api/v1/push":
+		return handleInsert(r, w)
+	case "/ready":
+		// See https://grafana.com/docs/loki/latest/api/#identify-ready-loki-instance
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte("ready"))
+		return true
+	default:
+		return false
+	}
+}
+
+// See https://grafana.com/docs/loki/latest/api/#push-log-entries-to-loki
+func handleInsert(r *http.Request, w http.ResponseWriter) bool {
+	contentType := r.Header.Get("Content-Type")
+	switch contentType {
+	case "application/json":
+		return handleJSON(r, w)
+	default:
+		// Protobuf request body should be handled by default according to https://grafana.com/docs/loki/latest/api/#push-log-entries-to-loki
+		return handleProtobuf(r, w)
+	}
+}
+
+func getCommonParams(r *http.Request) (*insertutils.CommonParams, error) {
+	cp, err := insertutils.GetCommonParams(r)
+	if err != nil {
+		return nil, err
+	}
+
+	// If parsed tenant is (0,0) it is likely to be default tenant
+	// Try parsing tenant from Loki headers
+	if cp.TenantID.AccountID == 0 && cp.TenantID.ProjectID == 0 {
+		org := r.Header.Get("X-Scope-OrgID")
+		if org != "" {
+			tenantID, err := logstorage.GetTenantIDFromString(org)
+			if err != nil {
+				return nil, err
+			}
+			cp.TenantID = tenantID
+		}
+
+	}
+
+	return cp, nil
+}

app/vlinsert/loki/loki_json.go

@@ -0,0 +1,205 @@
+package loki
+
+import (
+	"fmt"
+	"io"
+	"math"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
+	"github.com/VictoriaMetrics/metrics"
+	"github.com/valyala/fastjson"
+)
+
+var parserPool fastjson.ParserPool
+
+func handleJSON(r *http.Request, w http.ResponseWriter) bool {
+	startTime := time.Now()
+	lokiRequestsJSONTotal.Inc()
+	reader := r.Body
+	if r.Header.Get("Content-Encoding") == "gzip" {
+		zr, err := common.GetGzipReader(reader)
+		if err != nil {
+			httpserver.Errorf(w, r, "cannot initialize gzip reader: %s", err)
+			return true
+		}
+		defer common.PutGzipReader(zr)
+		reader = zr
+	}
+
+	wcr := writeconcurrencylimiter.GetReader(reader)
+	data, err := io.ReadAll(wcr)
+	writeconcurrencylimiter.PutReader(wcr)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot read request body: %s", err)
+		return true
+	}
+
+	cp, err := getCommonParams(r)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
+		return true
+	}
+	if err := vlstorage.CanWriteData(); err != nil {
+		httpserver.Errorf(w, r, "%s", err)
+		return true
+	}
+	lr := logstorage.GetLogRows(cp.StreamFields, cp.IgnoreFields)
+	processLogMessage := cp.GetProcessLogMessageFunc(lr)
+	n, err := parseJSONRequest(data, processLogMessage)
+	vlstorage.MustAddRows(lr)
+	logstorage.PutLogRows(lr)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot parse Loki json request: %s", err)
+		return true
+	}
+
+	rowsIngestedJSONTotal.Add(n)
+
+	// update lokiRequestJSONDuration only for successfully parsed requests
+	// There is no need in updating lokiRequestJSONDuration for request errors,
+	// since their timings are usually much smaller than the timing for successful request parsing.
+	lokiRequestJSONDuration.UpdateDuration(startTime)
+
+	return true
+}
+
+var (
+	lokiRequestsJSONTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="json"}`)
+	rowsIngestedJSONTotal   = metrics.NewCounter(`vl_rows_ingested_total{type="loki",format="json"}`)
+	lokiRequestJSONDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="json"}`)
+)
+
+func parseJSONRequest(data []byte, processLogMessage func(timestamp int64, fields []logstorage.Field)) (int, error) {
+	p := parserPool.Get()
+	defer parserPool.Put(p)
+	v, err := p.ParseBytes(data)
+	if err != nil {
+		return 0, fmt.Errorf("cannot parse JSON request body: %w", err)
+	}
+
+	streamsV := v.Get("streams")
+	if streamsV == nil {
+		return 0, fmt.Errorf("missing `streams` item in the parsed JSON: %q", v)
+	}
+	streams, err := streamsV.Array()
+	if err != nil {
+		return 0, fmt.Errorf("`streams` item in the parsed JSON must contain an array; got %q", streamsV)
+	}
+
+	currentTimestamp := time.Now().UnixNano()
+	var commonFields []logstorage.Field
+	rowsIngested := 0
+	for _, stream := range streams {
+		// populate common labels from `stream` dict
+		commonFields = commonFields[:0]
+		labelsV := stream.Get("stream")
+		var labels *fastjson.Object
+		if labelsV != nil {
+			o, err := labelsV.Object()
+			if err != nil {
+				return rowsIngested, fmt.Errorf("`stream` item in the parsed JSON must contain an object; got %q", labelsV)
+			}
+			labels = o
+		}
+		labels.Visit(func(k []byte, v *fastjson.Value) {
+			if err != nil {
+				return
+			}
+			vStr, errLocal := v.StringBytes()
+			if errLocal != nil {
+				err = fmt.Errorf("unexpected label value type for %q:%q; want string", k, v)
+				return
+			}
+			commonFields = append(commonFields, logstorage.Field{
+				Name:  bytesutil.ToUnsafeString(k),
+				Value: bytesutil.ToUnsafeString(vStr),
+			})
+		})
+		if err != nil {
+			return rowsIngested, fmt.Errorf("error when parsing `stream` object: %w", err)
+		}
+
+		// populate messages from `values` array
+		linesV := stream.Get("values")
+		if linesV == nil {
+			return rowsIngested, fmt.Errorf("missing `values` item in the parsed JSON %q", stream)
+		}
+		lines, err := linesV.Array()
+		if err != nil {
+			return rowsIngested, fmt.Errorf("`values` item in the parsed JSON must contain an array; got %q", linesV)
+		}
+
+		fields := commonFields
+		for _, line := range lines {
+			lineA, err := line.Array()
+			if err != nil {
+				return rowsIngested, fmt.Errorf("unexpected contents of `values` item; want array; got %q", line)
+			}
+			if len(lineA) != 2 {
+				return rowsIngested, fmt.Errorf("unexpected number of values in `values` item array %q; got %d want 2", line, len(lineA))
+			}
+
+			// parse timestamp
+			timestamp, err := lineA[0].StringBytes()
+			if err != nil {
+				return rowsIngested, fmt.Errorf("unexpected log timestamp type for %q; want string", lineA[0])
+			}
+			ts, err := parseLokiTimestamp(bytesutil.ToUnsafeString(timestamp))
+			if err != nil {
+				return rowsIngested, fmt.Errorf("cannot parse log timestamp %q: %w", timestamp, err)
+			}
+			if ts == 0 {
+				ts = currentTimestamp
+			}
+
+			// parse log message
+			msg, err := lineA[1].StringBytes()
+			if err != nil {
+				return rowsIngested, fmt.Errorf("unexpected log message type for %q; want string", lineA[1])
+			}
+
+			fields = append(fields[:len(commonFields)], logstorage.Field{
+				Name:  "_msg",
+				Value: bytesutil.ToUnsafeString(msg),
+			})
+			processLogMessage(ts, fields)
+		}
+		rowsIngested += len(lines)
+	}
+
+	return rowsIngested, nil
+}
+
+func parseLokiTimestamp(s string) (int64, error) {
+	if s == "" {
+		// Special case - an empty timestamp must be substituted with the current time by the caller.
+		return 0, nil
+	}
+	n, err := strconv.ParseInt(s, 10, 64)
+	if err != nil {
+		// Fall back to parsing floating-point value
+		f, err := strconv.ParseFloat(s, 64)
+		if err != nil {
+			return 0, err
+		}
+		if f > math.MaxInt64 {
+			return 0, fmt.Errorf("too big timestamp in nanoseconds: %v; mustn't exceed %v", f, int64(math.MaxInt64))
+		}
+		if f < math.MinInt64 {
+			return 0, fmt.Errorf("too small timestamp in nanoseconds: %v; must be bigger or equal to %v", f, int64(math.MinInt64))
+		}
+		n = int64(f)
+	}
+	if n < 0 {
+		return 0, fmt.Errorf("too small timestamp in nanoseconds: %d; must be bigger than 0", n)
+	}
+	return n, nil
+}

app/vlinsert/loki/loki_json_test.go

@@ -0,0 +1,130 @@
+package loki
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+)
+
+func TestParseJSONRequestFailure(t *testing.T) {
+	f := func(s string) {
+		t.Helper()
+		n, err := parseJSONRequest([]byte(s), func(_ int64, _ []logstorage.Field) {
+			t.Fatalf("unexpected call to parseJSONRequest callback!")
+		})
+		if err == nil {
+			t.Fatalf("expecting non-nil error")
+		}
+		if n != 0 {
+			t.Fatalf("unexpected number of parsed lines: %d; want 0", n)
+		}
+	}
+	f(``)
+
+	// Invalid json
+	f(`{}`)
+	f(`[]`)
+	f(`"foo"`)
+	f(`123`)
+
+	// invalid type for `streams` item
+	f(`{"streams":123}`)
+
+	// Missing `values` item
+	f(`{"streams":[{}]}`)
+
+	// Invalid type for `values` item
+	f(`{"streams":[{"values":"foobar"}]}`)
+
+	// Invalid type for `stream` item
+	f(`{"streams":[{"stream":[],"values":[]}]}`)
+
+	// Invalid type for `values` individual item
+	f(`{"streams":[{"values":[123]}]}`)
+
+	// Invalid length of `values` individual item
+	f(`{"streams":[{"values":[[]]}]}`)
+	f(`{"streams":[{"values":[["123"]]}]}`)
+	f(`{"streams":[{"values":[["123","456","789"]]}]}`)
+
+	// Invalid type for timestamp inside `values` individual item
+	f(`{"streams":[{"values":[[123,"456"]}]}`)
+
+	// Invalid type for log message
+	f(`{"streams":[{"values":[["123",1234]]}]}`)
+}
+
+func TestParseJSONRequestSuccess(t *testing.T) {
+	f := func(s string, resultExpected string) {
+		t.Helper()
+		var lines []string
+		n, err := parseJSONRequest([]byte(s), func(timestamp int64, fields []logstorage.Field) {
+			var a []string
+			for _, f := range fields {
+				a = append(a, f.String())
+			}
+			line := fmt.Sprintf("_time:%d %s", timestamp, strings.Join(a, " "))
+			lines = append(lines, line)
+		})
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if n != len(lines) {
+			t.Fatalf("unexpected number of lines parsed; got %d; want %d", n, len(lines))
+		}
+		result := strings.Join(lines, "\n")
+		if result != resultExpected {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, resultExpected)
+		}
+	}
+
+	// Empty streams
+	f(`{"streams":[]}`, ``)
+	f(`{"streams":[{"values":[]}]}`, ``)
+	f(`{"streams":[{"stream":{},"values":[]}]}`, ``)
+	f(`{"streams":[{"stream":{"foo":"bar"},"values":[]}]}`, ``)
+
+	// Empty stream labels
+	f(`{"streams":[{"values":[["1577836800000000001", "foo bar"]]}]}`, `_time:1577836800000000001 "_msg":"foo bar"`)
+	f(`{"streams":[{"stream":{},"values":[["1577836800000000001", "foo bar"]]}]}`, `_time:1577836800000000001 "_msg":"foo bar"`)
+
+	// Non-empty stream labels
+	f(`{"streams":[{"stream":{
+	"label1": "value1",
+	"label2": "value2"
+},"values":[
+	["1577836800000000001", "foo bar"],
+	["1477836900005000002", "abc"],
+	["147.78369e9", "foobar"]
+]}]}`, `_time:1577836800000000001 "label1":"value1" "label2":"value2" "_msg":"foo bar"
+_time:1477836900005000002 "label1":"value1" "label2":"value2" "_msg":"abc"
+_time:147783690000 "label1":"value1" "label2":"value2" "_msg":"foobar"`)
+
+	// Multiple streams
+	f(`{
+	"streams": [
+		{
+			"stream": {
+				"foo": "bar",
+				"a": "b"
+			},
+			"values": [
+				["1577836800000000001", "foo bar"],
+				["1577836900005000002", "abc"]
+			]
+		},
+		{
+			"stream": {
+				"x": "y"
+			},
+			"values": [
+				["1877836900005000002", "yx"]
+			]
+		}
+	]
+}`, `_time:1577836800000000001 "foo":"bar" "a":"b" "_msg":"foo bar"
+_time:1577836900005000002 "foo":"bar" "a":"b" "_msg":"abc"
+_time:1877836900005000002 "x":"y" "_msg":"yx"`)
+}

app/vlinsert/loki/loki_json_timing_test.go

@@ -0,0 +1,78 @@
+package loki
+
+import (
+	"fmt"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+)
+
+func BenchmarkParseJSONRequest(b *testing.B) {
+	for _, streams := range []int{5, 10} {
+		for _, rows := range []int{100, 1000} {
+			for _, labels := range []int{10, 50} {
+				b.Run(fmt.Sprintf("streams_%d/rows_%d/labels_%d", streams, rows, labels), func(b *testing.B) {
+					benchmarkParseJSONRequest(b, streams, rows, labels)
+				})
+			}
+		}
+	}
+}
+
+func benchmarkParseJSONRequest(b *testing.B, streams, rows, labels int) {
+	b.ReportAllocs()
+	b.SetBytes(int64(streams * rows))
+	b.RunParallel(func(pb *testing.PB) {
+		data := getJSONBody(streams, rows, labels)
+		for pb.Next() {
+			_, err := parseJSONRequest(data, func(_ int64, _ []logstorage.Field) {})
+			if err != nil {
+				panic(fmt.Errorf("unexpected error: %w", err))
+			}
+		}
+	})
+}
+
+func getJSONBody(streams, rows, labels int) []byte {
+	body := append([]byte{}, `{"streams":[`...)
+	now := time.Now().UnixNano()
+	valuePrefix := fmt.Sprintf(`["%d","value_`, now)
+
+	for i := 0; i < streams; i++ {
+		body = append(body, `{"stream":{`...)
+
+		for j := 0; j < labels; j++ {
+			body = append(body, `"label_`...)
+			body = strconv.AppendInt(body, int64(j), 10)
+			body = append(body, `":"value_`...)
+			body = strconv.AppendInt(body, int64(j), 10)
+			body = append(body, '"')
+			if j < labels-1 {
+				body = append(body, ',')
+			}
+
+		}
+		body = append(body, `}, "values":[`...)
+
+		for j := 0; j < rows; j++ {
+			body = append(body, valuePrefix...)
+			body = strconv.AppendInt(body, int64(j), 10)
+			body = append(body, `"]`...)
+			if j < rows-1 {
+				body = append(body, ',')
+			}
+		}
+
+		body = append(body, `]}`...)
+		if i < streams-1 {
+			body = append(body, ',')
+		}
+
+	}
+
+	body = append(body, `]}`...)
+
+	return body
+}

app/vlinsert/loki/loki_protobuf.go

@@ -0,0 +1,189 @@
+package loki
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
+	"github.com/VictoriaMetrics/metrics"
+	"github.com/golang/snappy"
+)
+
+var (
+	bytesBufPool bytesutil.ByteBufferPool
+	pushReqsPool sync.Pool
+)
+
+func handleProtobuf(r *http.Request, w http.ResponseWriter) bool {
+	startTime := time.Now()
+	lokiRequestsProtobufTotal.Inc()
+	wcr := writeconcurrencylimiter.GetReader(r.Body)
+	data, err := io.ReadAll(wcr)
+	writeconcurrencylimiter.PutReader(wcr)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot read request body: %s", err)
+		return true
+	}
+
+	cp, err := getCommonParams(r)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
+		return true
+	}
+	if err := vlstorage.CanWriteData(); err != nil {
+		httpserver.Errorf(w, r, "%s", err)
+		return true
+	}
+	lr := logstorage.GetLogRows(cp.StreamFields, cp.IgnoreFields)
+	processLogMessage := cp.GetProcessLogMessageFunc(lr)
+	n, err := parseProtobufRequest(data, processLogMessage)
+	vlstorage.MustAddRows(lr)
+	logstorage.PutLogRows(lr)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot parse Loki protobuf request: %s", err)
+		return true
+	}
+
+	rowsIngestedProtobufTotal.Add(n)
+
+	// update lokiRequestProtobufDuration only for successfully parsed requests
+	// There is no need in updating lokiRequestProtobufDuration for request errors,
+	// since their timings are usually much smaller than the timing for successful request parsing.
+	lokiRequestProtobufDuration.UpdateDuration(startTime)
+
+	return true
+}
+
+var (
+	lokiRequestsProtobufTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="protobuf"}`)
+	rowsIngestedProtobufTotal   = metrics.NewCounter(`vl_rows_ingested_total{type="loki",format="protobuf"}`)
+	lokiRequestProtobufDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="protobuf"}`)
+)
+
+func parseProtobufRequest(data []byte, processLogMessage func(timestamp int64, fields []logstorage.Field)) (int, error) {
+	bb := bytesBufPool.Get()
+	defer bytesBufPool.Put(bb)
+
+	buf, err := snappy.Decode(bb.B[:cap(bb.B)], data)
+	if err != nil {
+		return 0, fmt.Errorf("cannot decode snappy-encoded request body: %w", err)
+	}
+	bb.B = buf
+
+	req := getPushRequest()
+	defer putPushRequest(req)
+
+	err = req.Unmarshal(bb.B)
+	if err != nil {
+		return 0, fmt.Errorf("cannot parse request body: %w", err)
+	}
+
+	var commonFields []logstorage.Field
+	rowsIngested := 0
+	streams := req.Streams
+	currentTimestamp := time.Now().UnixNano()
+	for i := range streams {
+		stream := &streams[i]
+		// st.Labels contains labels for the stream.
+		// Labels are same for all entries in the stream.
+		commonFields, err = parsePromLabels(commonFields[:0], stream.Labels)
+		if err != nil {
+			return rowsIngested, fmt.Errorf("cannot parse stream labels %q: %w", stream.Labels, err)
+		}
+		fields := commonFields
+
+		entries := stream.Entries
+		for j := range entries {
+			entry := &entries[j]
+			fields = append(fields[:len(commonFields)], logstorage.Field{
+				Name:  "_msg",
+				Value: entry.Line,
+			})
+			ts := entry.Timestamp.UnixNano()
+			if ts == 0 {
+				ts = currentTimestamp
+			}
+			processLogMessage(ts, fields)
+		}
+		rowsIngested += len(stream.Entries)
+	}
+	return rowsIngested, nil
+}
+
+// parsePromLabels parses log fields in Prometheus text exposition format from s, appends them to dst and returns the result.
+//
+// See test data of promtail for examples: https://github.com/grafana/loki/blob/a24ef7b206e0ca63ee74ca6ecb0a09b745cd2258/pkg/push/types_test.go
+func parsePromLabels(dst []logstorage.Field, s string) ([]logstorage.Field, error) {
+	// Make sure s is wrapped into `{...}`
+	s = strings.TrimSpace(s)
+	if len(s) < 2 {
+		return nil, fmt.Errorf("too short string to parse: %q", s)
+	}
+	if s[0] != '{' {
+		return nil, fmt.Errorf("missing `{` at the beginning of %q", s)
+	}
+	if s[len(s)-1] != '}' {
+		return nil, fmt.Errorf("missing `}` at the end of %q", s)
+	}
+	s = s[1 : len(s)-1]
+
+	for len(s) > 0 {
+		// Parse label name
+		n := strings.IndexByte(s, '=')
+		if n < 0 {
+			return nil, fmt.Errorf("cannot find `=` char for label value at %s", s)
+		}
+		name := s[:n]
+		s = s[n+1:]
+
+		// Parse label value
+		qs, err := strconv.QuotedPrefix(s)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse value for label %q at %s: %w", name, s, err)
+		}
+		s = s[len(qs):]
+		value, err := strconv.Unquote(qs)
+		if err != nil {
+			return nil, fmt.Errorf("cannot unquote value %q for label %q: %w", qs, name, err)
+		}
+
+		// Append the found field to dst.
+		dst = append(dst, logstorage.Field{
+			Name:  name,
+			Value: value,
+		})
+
+		// Check whether there are other labels remaining
+		if len(s) == 0 {
+			break
+		}
+		if !strings.HasPrefix(s, ",") {
+			return nil, fmt.Errorf("missing `,` char at %s", s)
+		}
+		s = s[1:]
+		s = strings.TrimPrefix(s, " ")
+	}
+	return dst, nil
+}
+
+func getPushRequest() *PushRequest {
+	v := pushReqsPool.Get()
+	if v == nil {
+		return &PushRequest{}
+	}
+	return v.(*PushRequest)
+}
+
+func putPushRequest(req *PushRequest) {
+	req.Reset()
+	pushReqsPool.Put(req)
+}

app/vlinsert/loki/loki_protobuf_test.go

@@ -0,0 +1,171 @@
+package loki
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+	"github.com/golang/snappy"
+)
+
+func TestParseProtobufRequestSuccess(t *testing.T) {
+	f := func(s string, resultExpected string) {
+		t.Helper()
+		var pr PushRequest
+		n, err := parseJSONRequest([]byte(s), func(timestamp int64, fields []logstorage.Field) {
+			msg := ""
+			for _, f := range fields {
+				if f.Name == "_msg" {
+					msg = f.Value
+				}
+			}
+			var a []string
+			for _, f := range fields {
+				if f.Name == "_msg" {
+					continue
+				}
+				item := fmt.Sprintf("%s=%q", f.Name, f.Value)
+				a = append(a, item)
+			}
+			labels := "{" + strings.Join(a, ", ") + "}"
+			pr.Streams = append(pr.Streams, Stream{
+				Labels: labels,
+				Entries: []Entry{
+					{
+						Timestamp: time.Unix(0, timestamp),
+						Line:      msg,
+					},
+				},
+			})
+		})
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if n != len(pr.Streams) {
+			t.Fatalf("unexpected number of streams; got %d; want %d", len(pr.Streams), n)
+		}
+
+		data, err := pr.Marshal()
+		if err != nil {
+			t.Fatalf("unexpected error when marshaling PushRequest: %s", err)
+		}
+		encodedData := snappy.Encode(nil, data)
+
+		var lines []string
+		n, err = parseProtobufRequest(encodedData, func(timestamp int64, fields []logstorage.Field) {
+			var a []string
+			for _, f := range fields {
+				a = append(a, f.String())
+			}
+			line := fmt.Sprintf("_time:%d %s", timestamp, strings.Join(a, " "))
+			lines = append(lines, line)
+		})
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if n != len(lines) {
+			t.Fatalf("unexpected number of lines parsed; got %d; want %d", n, len(lines))
+		}
+		result := strings.Join(lines, "\n")
+		if result != resultExpected {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, resultExpected)
+		}
+	}
+
+	// Empty streams
+	f(`{"streams":[]}`, ``)
+	f(`{"streams":[{"values":[]}]}`, ``)
+	f(`{"streams":[{"stream":{},"values":[]}]}`, ``)
+	f(`{"streams":[{"stream":{"foo":"bar"},"values":[]}]}`, ``)
+
+	// Empty stream labels
+	f(`{"streams":[{"values":[["1577836800000000001", "foo bar"]]}]}`, `_time:1577836800000000001 "_msg":"foo bar"`)
+	f(`{"streams":[{"stream":{},"values":[["1577836800000000001", "foo bar"]]}]}`, `_time:1577836800000000001 "_msg":"foo bar"`)
+
+	// Non-empty stream labels
+	f(`{"streams":[{"stream":{
+	"label1": "value1",
+	"label2": "value2"
+},"values":[
+	["1577836800000000001", "foo bar"],
+	["1477836900005000002", "abc"],
+	["147.78369e9", "foobar"]
+]}]}`, `_time:1577836800000000001 "label1":"value1" "label2":"value2" "_msg":"foo bar"
+_time:1477836900005000002 "label1":"value1" "label2":"value2" "_msg":"abc"
+_time:147783690000 "label1":"value1" "label2":"value2" "_msg":"foobar"`)
+
+	// Multiple streams
+	f(`{
+	"streams": [
+		{
+			"stream": {
+				"foo": "bar",
+				"a": "b"
+			},
+			"values": [
+				["1577836800000000001", "foo bar"],
+				["1577836900005000002", "abc"]
+			]
+		},
+		{
+			"stream": {
+				"x": "y"
+			},
+			"values": [
+				["1877836900005000002", "yx"]
+			]
+		}
+	]
+}`, `_time:1577836800000000001 "foo":"bar" "a":"b" "_msg":"foo bar"
+_time:1577836900005000002 "foo":"bar" "a":"b" "_msg":"abc"
+_time:1877836900005000002 "x":"y" "_msg":"yx"`)
+}
+
+func TestParsePromLabelsSuccess(t *testing.T) {
+	f := func(s string) {
+		t.Helper()
+		fields, err := parsePromLabels(nil, s)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+
+		var a []string
+		for _, f := range fields {
+			a = append(a, fmt.Sprintf("%s=%q", f.Name, f.Value))
+		}
+		result := "{" + strings.Join(a, ", ") + "}"
+		if result != s {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, s)
+		}
+	}
+
+	f("{}")
+	f(`{foo="bar"}`)
+	f(`{foo="bar", baz="x", y="z"}`)
+	f(`{foo="ba\"r\\z\n", a="", b="\"\\"}`)
+}
+
+func TestParsePromLabelsFailure(t *testing.T) {
+	f := func(s string) {
+		t.Helper()
+		fields, err := parsePromLabels(nil, s)
+		if err == nil {
+			t.Fatalf("expecting non-nil error")
+		}
+		if len(fields) > 0 {
+			t.Fatalf("unexpected non-empty fields: %s", fields)
+		}
+	}
+
+	f("")
+	f("{")
+	f(`{foo}`)
+	f(`{foo=bar}`)
+	f(`{foo="bar}`)
+	f(`{foo="ba\",r}`)
+	f(`{foo="bar" baz="aa"}`)
+	f(`foobar`)
+	f(`foo{bar="baz"}`)
+}

app/vlinsert/loki/loki_protobuf_timing_test.go

@@ -0,0 +1,66 @@
+package loki
+
+import (
+	"fmt"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/golang/snappy"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+)
+
+func BenchmarkParseProtobufRequest(b *testing.B) {
+	for _, streams := range []int{5, 10} {
+		for _, rows := range []int{100, 1000} {
+			for _, labels := range []int{10, 50} {
+				b.Run(fmt.Sprintf("streams_%d/rows_%d/labels_%d", streams, rows, labels), func(b *testing.B) {
+					benchmarkParseProtobufRequest(b, streams, rows, labels)
+				})
+			}
+		}
+	}
+}
+
+func benchmarkParseProtobufRequest(b *testing.B, streams, rows, labels int) {
+	b.ReportAllocs()
+	b.SetBytes(int64(streams * rows))
+	b.RunParallel(func(pb *testing.PB) {
+		body := getProtobufBody(streams, rows, labels)
+		for pb.Next() {
+			_, err := parseProtobufRequest(body, func(_ int64, _ []logstorage.Field) {})
+			if err != nil {
+				panic(fmt.Errorf("unexpected error: %w", err))
+			}
+		}
+	})
+}
+
+func getProtobufBody(streams, rows, labels int) []byte {
+	var pr PushRequest
+
+	for i := 0; i < streams; i++ {
+		var st Stream
+
+		st.Labels = `{`
+		for j := 0; j < labels; j++ {
+			st.Labels += `label_` + strconv.Itoa(j) + `="value_` + strconv.Itoa(j) + `"`
+			if j < labels-1 {
+				st.Labels += `,`
+			}
+		}
+		st.Labels += `}`
+
+		for j := 0; j < rows; j++ {
+			st.Entries = append(st.Entries, Entry{Timestamp: time.Now(), Line: "value_" + strconv.Itoa(j)})
+		}
+
+		pr.Streams = append(pr.Streams, st)
+	}
+
+	body, _ := pr.Marshal()
+	encodedBody := snappy.Encode(nil, body)
+
+	return encodedBody
+}

app/vlinsert/loki/push_request.proto

@@ -0,0 +1,38 @@
+syntax = "proto3";
+
+// source: https://raw.githubusercontent.com/grafana/loki/main/pkg/push/push.proto
+// Licensed under the Apache License, Version 2.0 (the "License");
+// https://github.com/grafana/loki/blob/main/pkg/push/LICENSE
+
+package logproto;
+
+import "gogoproto/gogo.proto";
+import "google/protobuf/timestamp.proto";
+
+option go_package = "github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/loki";
+
+message PushRequest {
+  repeated StreamAdapter streams = 1 [
+    (gogoproto.jsontag) = "streams",
+    (gogoproto.customtype) = "Stream"
+  ];
+}
+
+message StreamAdapter {
+  string labels = 1 [(gogoproto.jsontag) = "labels"];
+  repeated EntryAdapter entries = 2 [
+    (gogoproto.nullable) = false,
+    (gogoproto.jsontag) = "entries"
+  ];
+  // hash contains the original hash of the stream.
+  uint64 hash = 3 [(gogoproto.jsontag) = "-"];
+}
+
+message EntryAdapter {
+  google.protobuf.Timestamp timestamp = 1 [
+    (gogoproto.stdtime) = true,
+    (gogoproto.nullable) = false,
+    (gogoproto.jsontag) = "ts"
+  ];
+  string line = 2 [(gogoproto.jsontag) = "line"];
+}

app/vlinsert/loki/timestamp.go

@@ -0,0 +1,110 @@
+package loki
+
+// source: https://raw.githubusercontent.com/grafana/loki/main/pkg/push/timestamp.go
+// Licensed under the Apache License, Version 2.0 (the "License");
+// https://github.com/grafana/loki/blob/main/pkg/push/LICENSE
+
+import (
+	"errors"
+	"strconv"
+	"time"
+
+	"github.com/gogo/protobuf/types"
+)
+
+const (
+	// Seconds field of the earliest valid Timestamp.
+	// This is time.Date(1, 1, 1, 0, 0, 0, 0, time.UTC).Unix().
+	minValidSeconds = -62135596800
+	// Seconds field just after the latest valid Timestamp.
+	// This is time.Date(10000, 1, 1, 0, 0, 0, 0, time.UTC).Unix().
+	maxValidSeconds = 253402300800
+)
+
+// validateTimestamp determines whether a Timestamp is valid.
+// A valid timestamp represents a time in the range
+// [0001-01-01, 10000-01-01) and has a Nanos field
+// in the range [0, 1e9).
+//
+// If the Timestamp is valid, validateTimestamp returns nil.
+// Otherwise, it returns an error that describes
+// the problem.
+//
+// Every valid Timestamp can be represented by a time.Time, but the converse is not true.
+func validateTimestamp(ts *types.Timestamp) error {
+	if ts == nil {
+		return errors.New("timestamp: nil Timestamp")
+	}
+	if ts.Seconds < minValidSeconds {
+		return errors.New("timestamp: " + formatTimestamp(ts) + " before 0001-01-01")
+	}
+	if ts.Seconds >= maxValidSeconds {
+		return errors.New("timestamp: " + formatTimestamp(ts) + " after 10000-01-01")
+	}
+	if ts.Nanos < 0 || ts.Nanos >= 1e9 {
+		return errors.New("timestamp: " + formatTimestamp(ts) + ": nanos not in range [0, 1e9)")
+	}
+	return nil
+}
+
+// formatTimestamp is equivalent to fmt.Sprintf("%#v", ts)
+// but avoids the escape incurred by using fmt.Sprintf, eliminating
+// unnecessary heap allocations.
+func formatTimestamp(ts *types.Timestamp) string {
+	if ts == nil {
+		return "nil"
+	}
+
+	seconds := strconv.FormatInt(ts.Seconds, 10)
+	nanos := strconv.FormatInt(int64(ts.Nanos), 10)
+	return "&types.Timestamp{Seconds: " + seconds + ",\nNanos: " + nanos + ",\n}"
+}
+
+func sizeOfStdTime(t time.Time) int {
+	ts, err := timestampProto(t)
+	if err != nil {
+		return 0
+	}
+	return ts.Size()
+}
+
+func stdTimeMarshalTo(t time.Time, data []byte) (int, error) {
+	ts, err := timestampProto(t)
+	if err != nil {
+		return 0, err
+	}
+	return ts.MarshalTo(data)
+}
+
+func stdTimeUnmarshal(t *time.Time, data []byte) error {
+	ts := &types.Timestamp{}
+	if err := ts.Unmarshal(data); err != nil {
+		return err
+	}
+	tt, err := timestampFromProto(ts)
+	if err != nil {
+		return err
+	}
+	*t = tt
+	return nil
+}
+
+func timestampFromProto(ts *types.Timestamp) (time.Time, error) {
+	// Don't return the zero value on error, because corresponds to a valid
+	// timestamp. Instead return whatever time.Unix gives us.
+	var t time.Time
+	if ts == nil {
+		t = time.Unix(0, 0).UTC() // treat nil like the empty Timestamp
+	} else {
+		t = time.Unix(ts.Seconds, int64(ts.Nanos)).UTC()
+	}
+	return t, validateTimestamp(ts)
+}
+
+func timestampProto(t time.Time) (types.Timestamp, error) {
+	ts := types.Timestamp{
+		Seconds: t.Unix(),
+		Nanos:   int32(t.Nanosecond()),
+	}
+	return ts, validateTimestamp(&ts)
+}

app/vlinsert/loki/types.go

@@ -0,0 +1,481 @@
+package loki
+
+// source: https://raw.githubusercontent.com/grafana/loki/main/pkg/push/types.go
+// Licensed under the Apache License, Version 2.0 (the "License");
+// https://github.com/grafana/loki/blob/main/pkg/push/LICENSE
+
+import (
+	"fmt"
+	"io"
+	"time"
+)
+
+// Stream contains a unique labels set as a string and a set of entries for it.
+// We are not using the proto generated version but this custom one so that we
+// can improve serialization see benchmark.
+type Stream struct {
+	Labels  string  `protobuf:"bytes,1,opt,name=labels,proto3" json:"labels"`
+	Entries []Entry `protobuf:"bytes,2,rep,name=entries,proto3,customtype=EntryAdapter" json:"entries"`
+	Hash    uint64  `protobuf:"varint,3,opt,name=hash,proto3" json:"-"`
+}
+
+// Entry is a log entry with a timestamp.
+type Entry struct {
+	Timestamp time.Time `protobuf:"bytes,1,opt,name=timestamp,proto3,stdtime" json:"ts"`
+	Line      string    `protobuf:"bytes,2,opt,name=line,proto3" json:"line"`
+}
+
+// Marshal implements the proto.Marshaler interface.
+func (m *Stream) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+// MarshalTo marshals m to dst.
+func (m *Stream) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+// MarshalToSizedBuffer marshals m to the sized buffer.
+func (m *Stream) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.Hash != 0 {
+		i = encodeVarintPush(dAtA, i, m.Hash)
+		i--
+		dAtA[i] = 0x18
+	}
+	if len(m.Entries) > 0 {
+		for iNdEx := len(m.Entries) - 1; iNdEx >= 0; iNdEx-- {
+			{
+				size, err := m.Entries[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintPush(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x12
+		}
+	}
+	if len(m.Labels) > 0 {
+		i -= len(m.Labels)
+		copy(dAtA[i:], m.Labels)
+		i = encodeVarintPush(dAtA, i, uint64(len(m.Labels)))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
+// Marshal implements the proto.Marshaler interface.
+func (m *Entry) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+// MarshalTo marshals m to dst.
+func (m *Entry) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+// MarshalToSizedBuffer marshals m to the sized buffer.
+func (m *Entry) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if len(m.Line) > 0 {
+		i -= len(m.Line)
+		copy(dAtA[i:], m.Line)
+		i = encodeVarintPush(dAtA, i, uint64(len(m.Line)))
+		i--
+		dAtA[i] = 0x12
+	}
+	n7, err7 := stdTimeMarshalTo(m.Timestamp, dAtA[i-sizeOfStdTime(m.Timestamp):])
+	if err7 != nil {
+		return 0, err7
+	}
+	i -= n7
+	i = encodeVarintPush(dAtA, i, uint64(n7))
+	i--
+	dAtA[i] = 0xa
+	return len(dAtA) - i, nil
+}
+
+// Unmarshal unmarshals the given data into m.
+func (m *Stream) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowPush
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: StreamAdapter: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: StreamAdapter: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Labels", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPush
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthPush
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthPush
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Labels = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Entries", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPush
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthPush
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthPush
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Entries = append(m.Entries, Entry{})
+			if err := m.Entries[len(m.Entries)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Hash", wireType)
+			}
+			m.Hash = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPush
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Hash |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		default:
+			iNdEx = preIndex
+			skippy, err := skipPush(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthPush
+			}
+			if (iNdEx + skippy) < 0 {
+				return ErrInvalidLengthPush
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+
+// Unmarshal unmarshals the given data into m.
+func (m *Entry) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowPush
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: EntryAdapter: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: EntryAdapter: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Timestamp", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPush
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthPush
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthPush
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := stdTimeUnmarshal(&m.Timestamp, dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Line", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPush
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthPush
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthPush
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Line = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipPush(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthPush
+			}
+			if (iNdEx + skippy) < 0 {
+				return ErrInvalidLengthPush
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+
+// Size returns the size of the serialized Stream.
+func (m *Stream) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.Labels)
+	if l > 0 {
+		n += 1 + l + sovPush(uint64(l))
+	}
+	if len(m.Entries) > 0 {
+		for _, e := range m.Entries {
+			l = e.Size()
+			n += 1 + l + sovPush(uint64(l))
+		}
+	}
+	if m.Hash != 0 {
+		n += 1 + sovPush(m.Hash)
+	}
+	return n
+}
+
+// Size returns the size of the serialized Entry
+func (m *Entry) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = sizeOfStdTime(m.Timestamp)
+	n += 1 + l + sovPush(uint64(l))
+	l = len(m.Line)
+	if l > 0 {
+		n += 1 + l + sovPush(uint64(l))
+	}
+	return n
+}
+
+// Equal returns true if the two Streams are equal.
+func (m *Stream) Equal(that interface{}) bool {
+	if that == nil {
+		return m == nil
+	}
+
+	that1, ok := that.(*Stream)
+	if !ok {
+		that2, ok := that.(Stream)
+		if ok {
+			that1 = &that2
+		} else {
+			return false
+		}
+	}
+	if that1 == nil {
+		return m == nil
+	} else if m == nil {
+		return false
+	}
+	if m.Labels != that1.Labels {
+		return false
+	}
+	if len(m.Entries) != len(that1.Entries) {
+		return false
+	}
+	for i := range m.Entries {
+		if !m.Entries[i].Equal(that1.Entries[i]) {
+			return false
+		}
+	}
+	return m.Hash == that1.Hash
+}
+
+// Equal returns true if the two Entries are equal.
+func (m *Entry) Equal(that interface{}) bool {
+	if that == nil {
+		return m == nil
+	}
+
+	that1, ok := that.(*Entry)
+	if !ok {
+		that2, ok := that.(Entry)
+		if ok {
+			that1 = &that2
+		} else {
+			return false
+		}
+	}
+	if that1 == nil {
+		return m == nil
+	} else if m == nil {
+		return false
+	}
+	if !m.Timestamp.Equal(that1.Timestamp) {
+		return false
+	}
+	if m.Line != that1.Line {
+		return false
+	}
+	return true
+}

app/vlinsert/main.go

@@ -0,0 +1,43 @@
+package vlinsert
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/elasticsearch"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/jsonline"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/loki"
+)
+
+// Init initializes vlinsert
+func Init() {
+}
+
+// Stop stops vlinsert
+func Stop() {
+}
+
+// RequestHandler handles insert requests for VictoriaLogs
+func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
+	path := r.URL.Path
+	if !strings.HasPrefix(path, "/insert/") {
+		// Skip requests, which do not start with /insert/, since these aren't our requests.
+		return false
+	}
+	path = strings.TrimPrefix(path, "/insert")
+	path = strings.ReplaceAll(path, "//", "/")
+
+	if path == "/jsonline" {
+		return jsonline.RequestHandler(w, r)
+	}
+	switch {
+	case strings.HasPrefix(path, "/elasticsearch/"):
+		path = strings.TrimPrefix(path, "/elasticsearch")
+		return elasticsearch.RequestHandler(path, w, r)
+	case strings.HasPrefix(path, "/loki/"):
+		path = strings.TrimPrefix(path, "/loki")
+		return loki.RequestHandler(path, w, r)
+	default:
+		return false
+	}
+}

app/vlogscli/README.md

@@ -1 +0,0 @@
-VictoriaLogs source code has been moved to [github.com/VictoriaMetrics/VictoriaLogs](https://github.com/VictoriaMetrics/VictoriaLogs/).

app/vlogsgenerator/README.md

@@ -1 +0,0 @@
-VictoriaLogs source code has been moved to [github.com/VictoriaMetrics/VictoriaLogs](https://github.com/VictoriaMetrics/VictoriaLogs/).

app/vlselect/README.md

@@ -1 +0,0 @@
-VictoriaLogs source code has been moved to [github.com/VictoriaMetrics/VictoriaLogs](https://github.com/VictoriaMetrics/VictoriaLogs/).

app/vlselect/logsql/logsql.go

@@ -0,0 +1,87 @@
+package logsql
+
+import (
+	"net/http"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+)
+
+var (
+	maxSortBufferSize = flagutil.NewBytes("select.maxSortBufferSize", 1024*1024, "Query results from /select/logsql/query are automatically sorted by _time "+
+		"if their summary size doesn't exceed this value; otherwise, query results are streamed in the response without sorting; "+
+		"too big value for this flag may result in high memory usage since the sorting is performed in memory")
+)
+
+// ProcessQueryRequest handles /select/logsql/query request
+func ProcessQueryRequest(w http.ResponseWriter, r *http.Request, stopCh <-chan struct{}, cancel func()) {
+	// Extract tenantID
+	tenantID, err := logstorage.GetTenantIDFromRequest(r)
+	if err != nil {
+		httpserver.Errorf(w, r, "%s", err)
+		return
+	}
+	limit, err := httputils.GetInt(r, "limit")
+	if err != nil {
+		httpserver.Errorf(w, r, "%s", err)
+		return
+	}
+
+	qStr := r.FormValue("query")
+	q, err := logstorage.ParseQuery(qStr)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot parse query [%s]: %s", qStr, err)
+		return
+	}
+	w.Header().Set("Content-Type", "application/stream+json; charset=utf-8")
+
+	sw := getSortWriter()
+	sw.Init(w, maxSortBufferSize.IntN(), limit)
+	tenantIDs := []logstorage.TenantID{tenantID}
+	vlstorage.RunQuery(tenantIDs, q, stopCh, func(columns []logstorage.BlockColumn) {
+		if len(columns) == 0 {
+			return
+		}
+		rowsCount := len(columns[0].Values)
+
+		// skip entries with empty _stream column
+		// _stream is empty in case indexdb entry was not flushed to the storage yet
+		// skipping such entries makes the result more consistent
+		streamCol := 0
+
+		// fast path
+		// _stream column is a built-in column and it is always supposed to be at the same position
+		if len(columns) >= 2 && columns[1].Name == "_stream" {
+			streamCol = 1
+		} else {
+			for i := 1; i < len(columns); i++ {
+				if columns[i].Name == "_stream" {
+					streamCol = i
+					break
+				}
+			}
+		}
+
+		bb := blockResultPool.Get()
+		for rowIdx := 0; rowIdx < rowsCount; rowIdx++ {
+			if columns[streamCol].Values[rowIdx] == "" {
+				continue
+			}
+			WriteJSONRow(bb, columns, rowIdx)
+		}
+
+		if !sw.TryWrite(bb.B) {
+			cancel()
+		}
+
+		blockResultPool.Put(bb)
+	})
+	sw.FinalFlush()
+	putSortWriter(sw)
+}
+
+var blockResultPool bytesutil.ByteBufferPool

app/vlselect/logsql/query_response.qtpl

@@ -0,0 +1,41 @@
+{% import (
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+) %}
+
+{% stripspace %}
+
+// JSONRow creates JSON row from the given fields.
+{% func JSONRow(columns []logstorage.BlockColumn, rowIdx int) %}
+{
+	{% code c := &columns[0] %}
+	{%q= c.Name %}:{%q= c.Values[rowIdx] %}
+	{% code columns = columns[1:] %}
+	{% for colIdx := range columns %}
+		{% code c := &columns[colIdx] %}
+		,{%q= c.Name %}:{%q= c.Values[rowIdx] %}
+	{% endfor %}
+}{% newline %}
+{% endfunc %}
+
+// JSONRows prints formatted rows
+{% func JSONRows(rows [][]logstorage.Field) %}
+	{% if len(rows) == 0 %}
+		{% return %}
+	{% endif %}
+	{% for _, fields := range rows %}
+	{
+		{% if len(fields) > 0 %}
+			{% code
+				f := fields[0]
+				fields = fields[1:]
+			%}
+			{%q= f.Name %}:{%q= f.Value %}
+			{% for _, f := range fields %}
+				,{%q= f.Name %}:{%q= f.Value %}
+			{% endfor %}
+		{% endif %}
+	}{% newline %}
+	{% endfor %}
+{% endfunc %}
+
+{% endstripspace %}

app/vlselect/logsql/query_response.qtpl.go

@@ -0,0 +1,166 @@
+// Code generated by qtc from "query_response.qtpl". DO NOT EDIT.
+// See https://github.com/valyala/quicktemplate for details.
+
+//line app/vlselect/logsql/query_response.qtpl:1
+package logsql
+
+//line app/vlselect/logsql/query_response.qtpl:1
+import (
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+)
+
+// JSONRow creates JSON row from the given fields.
+
+//line app/vlselect/logsql/query_response.qtpl:8
+import (
+	qtio422016 "io"
+
+	qt422016 "github.com/valyala/quicktemplate"
+)
+
+//line app/vlselect/logsql/query_response.qtpl:8
+var (
+	_ = qtio422016.Copy
+	_ = qt422016.AcquireByteBuffer
+)
+
+//line app/vlselect/logsql/query_response.qtpl:8
+func StreamJSONRow(qw422016 *qt422016.Writer, columns []logstorage.BlockColumn, rowIdx int) {
+//line app/vlselect/logsql/query_response.qtpl:8
+	qw422016.N().S(`{`)
+//line app/vlselect/logsql/query_response.qtpl:10
+	c := &columns[0]
+
+//line app/vlselect/logsql/query_response.qtpl:11
+	qw422016.N().Q(c.Name)
+//line app/vlselect/logsql/query_response.qtpl:11
+	qw422016.N().S(`:`)
+//line app/vlselect/logsql/query_response.qtpl:11
+	qw422016.N().Q(c.Values[rowIdx])
+//line app/vlselect/logsql/query_response.qtpl:12
+	columns = columns[1:]
+
+//line app/vlselect/logsql/query_response.qtpl:13
+	for colIdx := range columns {
+//line app/vlselect/logsql/query_response.qtpl:14
+		c := &columns[colIdx]
+
+//line app/vlselect/logsql/query_response.qtpl:14
+		qw422016.N().S(`,`)
+//line app/vlselect/logsql/query_response.qtpl:15
+		qw422016.N().Q(c.Name)
+//line app/vlselect/logsql/query_response.qtpl:15
+		qw422016.N().S(`:`)
+//line app/vlselect/logsql/query_response.qtpl:15
+		qw422016.N().Q(c.Values[rowIdx])
+//line app/vlselect/logsql/query_response.qtpl:16
+	}
+//line app/vlselect/logsql/query_response.qtpl:16
+	qw422016.N().S(`}`)
+//line app/vlselect/logsql/query_response.qtpl:17
+	qw422016.N().S(`
+`)
+//line app/vlselect/logsql/query_response.qtpl:18
+}
+
+//line app/vlselect/logsql/query_response.qtpl:18
+func WriteJSONRow(qq422016 qtio422016.Writer, columns []logstorage.BlockColumn, rowIdx int) {
+//line app/vlselect/logsql/query_response.qtpl:18
+	qw422016 := qt422016.AcquireWriter(qq422016)
+//line app/vlselect/logsql/query_response.qtpl:18
+	StreamJSONRow(qw422016, columns, rowIdx)
+//line app/vlselect/logsql/query_response.qtpl:18
+	qt422016.ReleaseWriter(qw422016)
+//line app/vlselect/logsql/query_response.qtpl:18
+}
+
+//line app/vlselect/logsql/query_response.qtpl:18
+func JSONRow(columns []logstorage.BlockColumn, rowIdx int) string {
+//line app/vlselect/logsql/query_response.qtpl:18
+	qb422016 := qt422016.AcquireByteBuffer()
+//line app/vlselect/logsql/query_response.qtpl:18
+	WriteJSONRow(qb422016, columns, rowIdx)
+//line app/vlselect/logsql/query_response.qtpl:18
+	qs422016 := string(qb422016.B)
+//line app/vlselect/logsql/query_response.qtpl:18
+	qt422016.ReleaseByteBuffer(qb422016)
+//line app/vlselect/logsql/query_response.qtpl:18
+	return qs422016
+//line app/vlselect/logsql/query_response.qtpl:18
+}
+
+// JSONRows prints formatted rows
+
+//line app/vlselect/logsql/query_response.qtpl:21
+func StreamJSONRows(qw422016 *qt422016.Writer, rows [][]logstorage.Field) {
+//line app/vlselect/logsql/query_response.qtpl:22
+	if len(rows) == 0 {
+//line app/vlselect/logsql/query_response.qtpl:23
+		return
+//line app/vlselect/logsql/query_response.qtpl:24
+	}
+//line app/vlselect/logsql/query_response.qtpl:25
+	for _, fields := range rows {
+//line app/vlselect/logsql/query_response.qtpl:25
+		qw422016.N().S(`{`)
+//line app/vlselect/logsql/query_response.qtpl:27
+		if len(fields) > 0 {
+//line app/vlselect/logsql/query_response.qtpl:29
+			f := fields[0]
+			fields = fields[1:]
+
+//line app/vlselect/logsql/query_response.qtpl:32
+			qw422016.N().Q(f.Name)
+//line app/vlselect/logsql/query_response.qtpl:32
+			qw422016.N().S(`:`)
+//line app/vlselect/logsql/query_response.qtpl:32
+			qw422016.N().Q(f.Value)
+//line app/vlselect/logsql/query_response.qtpl:33
+			for _, f := range fields {
+//line app/vlselect/logsql/query_response.qtpl:33
+				qw422016.N().S(`,`)
+//line app/vlselect/logsql/query_response.qtpl:34
+				qw422016.N().Q(f.Name)
+//line app/vlselect/logsql/query_response.qtpl:34
+				qw422016.N().S(`:`)
+//line app/vlselect/logsql/query_response.qtpl:34
+				qw422016.N().Q(f.Value)
+//line app/vlselect/logsql/query_response.qtpl:35
+			}
+//line app/vlselect/logsql/query_response.qtpl:36
+		}
+//line app/vlselect/logsql/query_response.qtpl:36
+		qw422016.N().S(`}`)
+//line app/vlselect/logsql/query_response.qtpl:37
+		qw422016.N().S(`
+`)
+//line app/vlselect/logsql/query_response.qtpl:38
+	}
+//line app/vlselect/logsql/query_response.qtpl:39
+}
+
+//line app/vlselect/logsql/query_response.qtpl:39
+func WriteJSONRows(qq422016 qtio422016.Writer, rows [][]logstorage.Field) {
+//line app/vlselect/logsql/query_response.qtpl:39
+	qw422016 := qt422016.AcquireWriter(qq422016)
+//line app/vlselect/logsql/query_response.qtpl:39
+	StreamJSONRows(qw422016, rows)
+//line app/vlselect/logsql/query_response.qtpl:39
+	qt422016.ReleaseWriter(qw422016)
+//line app/vlselect/logsql/query_response.qtpl:39
+}
+
+//line app/vlselect/logsql/query_response.qtpl:39
+func JSONRows(rows [][]logstorage.Field) string {
+//line app/vlselect/logsql/query_response.qtpl:39
+	qb422016 := qt422016.AcquireByteBuffer()
+//line app/vlselect/logsql/query_response.qtpl:39
+	WriteJSONRows(qb422016, rows)
+//line app/vlselect/logsql/query_response.qtpl:39
+	qs422016 := string(qb422016.B)
+//line app/vlselect/logsql/query_response.qtpl:39
+	qt422016.ReleaseByteBuffer(qb422016)
+//line app/vlselect/logsql/query_response.qtpl:39
+	return qs422016
+//line app/vlselect/logsql/query_response.qtpl:39
+}

app/vlselect/logsql/sort_writer.go

@@ -0,0 +1,290 @@
+package logsql
+
+import (
+	"bytes"
+	"io"
+	"sort"
+	"sync"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logjson"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+)
+
+func getSortWriter() *sortWriter {
+	v := sortWriterPool.Get()
+	if v == nil {
+		return &sortWriter{}
+	}
+	return v.(*sortWriter)
+}
+
+func putSortWriter(sw *sortWriter) {
+	sw.reset()
+	sortWriterPool.Put(sw)
+}
+
+var sortWriterPool sync.Pool
+
+// sortWriter expects JSON line stream to be written to it.
+//
+// It buffers the incoming data until its size reaches maxBufLen.
+// Then it streams the buffered data and all the incoming data to w.
+//
+// The FinalFlush() must be called when all the data is written.
+// If the buf isn't empty at FinalFlush() call, then the buffered data
+// is sorted by _time field.
+type sortWriter struct {
+	mu sync.Mutex
+	w  io.Writer
+
+	maxLines     int
+	linesWritten int
+
+	maxBufLen  int
+	buf        []byte
+	bufFlushed bool
+
+	hasErr bool
+}
+
+func (sw *sortWriter) reset() {
+	sw.w = nil
+
+	sw.maxLines = 0
+	sw.linesWritten = 0
+
+	sw.maxBufLen = 0
+	sw.buf = sw.buf[:0]
+	sw.bufFlushed = false
+	sw.hasErr = false
+}
+
+// Init initializes sw.
+//
+// If maxLines is set to positive value, then sw accepts up to maxLines
+// and then rejects all the other lines by returning false from TryWrite.
+func (sw *sortWriter) Init(w io.Writer, maxBufLen, maxLines int) {
+	sw.reset()
+
+	sw.w = w
+	sw.maxBufLen = maxBufLen
+	sw.maxLines = maxLines
+}
+
+// TryWrite writes p to sw.
+//
+// True is returned on successful write, false otherwise.
+//
+// Unsuccessful write may occur on underlying write error or when maxLines lines are already written to sw.
+func (sw *sortWriter) TryWrite(p []byte) bool {
+	sw.mu.Lock()
+	defer sw.mu.Unlock()
+
+	if sw.hasErr {
+		return false
+	}
+
+	if sw.bufFlushed {
+		if !sw.writeToUnderlyingWriterLocked(p) {
+			sw.hasErr = true
+			return false
+		}
+		return true
+	}
+
+	if len(sw.buf)+len(p) < sw.maxBufLen {
+		sw.buf = append(sw.buf, p...)
+		return true
+	}
+
+	sw.bufFlushed = true
+	if !sw.writeToUnderlyingWriterLocked(sw.buf) {
+		sw.hasErr = true
+		return false
+	}
+	sw.buf = sw.buf[:0]
+
+	if !sw.writeToUnderlyingWriterLocked(p) {
+		sw.hasErr = true
+		return false
+	}
+	return true
+}
+
+func (sw *sortWriter) writeToUnderlyingWriterLocked(p []byte) bool {
+	if len(p) == 0 {
+		return true
+	}
+	if sw.maxLines > 0 {
+		if sw.linesWritten >= sw.maxLines {
+			return false
+		}
+		var linesLeft int
+		p, linesLeft = trimLines(p, sw.maxLines-sw.linesWritten)
+		sw.linesWritten += linesLeft
+	}
+	if _, err := sw.w.Write(p); err != nil {
+		return false
+	}
+	return true
+}
+
+func trimLines(p []byte, maxLines int) ([]byte, int) {
+	if maxLines <= 0 {
+		return nil, 0
+	}
+	n := bytes.Count(p, newline)
+	if n < maxLines {
+		return p, n
+	}
+	for n >= maxLines {
+		idx := bytes.LastIndexByte(p, '\n')
+		p = p[:idx]
+		n--
+	}
+	return p[:len(p)+1], maxLines
+}
+
+var newline = []byte("\n")
+
+func (sw *sortWriter) FinalFlush() {
+	if sw.hasErr || sw.bufFlushed {
+		return
+	}
+
+	rs := getRowsSorter()
+	rs.parseRows(sw.buf)
+	rs.sort()
+
+	rows := rs.rows
+	if sw.maxLines > 0 && len(rows) > sw.maxLines {
+		rows = rows[:sw.maxLines]
+	}
+	WriteJSONRows(sw.w, rows)
+
+	putRowsSorter(rs)
+}
+
+func getRowsSorter() *rowsSorter {
+	v := rowsSorterPool.Get()
+	if v == nil {
+		return &rowsSorter{}
+	}
+	return v.(*rowsSorter)
+}
+
+func putRowsSorter(rs *rowsSorter) {
+	rs.reset()
+	rowsSorterPool.Put(rs)
+}
+
+var rowsSorterPool sync.Pool
+
+type rowsSorter struct {
+	buf       []byte
+	fieldsBuf []logstorage.Field
+	rows      [][]logstorage.Field
+	times     []string
+}
+
+func (rs *rowsSorter) reset() {
+	rs.buf = rs.buf[:0]
+
+	fieldsBuf := rs.fieldsBuf
+	for i := range fieldsBuf {
+		fieldsBuf[i].Reset()
+	}
+	rs.fieldsBuf = fieldsBuf[:0]
+
+	rows := rs.rows
+	for i := range rows {
+		rows[i] = nil
+	}
+	rs.rows = rows[:0]
+
+	times := rs.times
+	for i := range times {
+		times[i] = ""
+	}
+	rs.times = times[:0]
+}
+
+func (rs *rowsSorter) parseRows(src []byte) {
+	rs.reset()
+
+	buf := rs.buf
+	fieldsBuf := rs.fieldsBuf
+	rows := rs.rows
+	times := rs.times
+
+	p := logjson.GetParser()
+	for len(src) > 0 {
+		var line []byte
+		n := bytes.IndexByte(src, '\n')
+		if n < 0 {
+			line = src
+			src = nil
+		} else {
+			line = src[:n]
+			src = src[n+1:]
+		}
+		if len(line) == 0 {
+			continue
+		}
+
+		if err := p.ParseLogMessage(line); err != nil {
+			logger.Panicf("BUG: unexpected invalid JSON line: %s", err)
+		}
+
+		timeValue := ""
+		fieldsBufLen := len(fieldsBuf)
+		for _, f := range p.Fields {
+			bufLen := len(buf)
+			buf = append(buf, f.Name...)
+			name := bytesutil.ToUnsafeString(buf[bufLen:])
+
+			bufLen = len(buf)
+			buf = append(buf, f.Value...)
+			value := bytesutil.ToUnsafeString(buf[bufLen:])
+
+			fieldsBuf = append(fieldsBuf, logstorage.Field{
+				Name:  name,
+				Value: value,
+			})
+
+			if name == "_time" {
+				timeValue = value
+			}
+		}
+		rows = append(rows, fieldsBuf[fieldsBufLen:])
+		times = append(times, timeValue)
+	}
+	logjson.PutParser(p)
+
+	rs.buf = buf
+	rs.fieldsBuf = fieldsBuf
+	rs.rows = rows
+	rs.times = times
+}
+
+func (rs *rowsSorter) Len() int {
+	return len(rs.rows)
+}
+
+func (rs *rowsSorter) Less(i, j int) bool {
+	times := rs.times
+	return times[i] < times[j]
+}
+
+func (rs *rowsSorter) Swap(i, j int) {
+	times := rs.times
+	rows := rs.rows
+	times[i], times[j] = times[j], times[i]
+	rows[i], rows[j] = rows[j], rows[i]
+}
+
+func (rs *rowsSorter) sort() {
+	sort.Sort(rs)
+}

app/vlselect/logsql/sort_writer_test.go

@@ -0,0 +1,46 @@
+package logsql
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+)
+
+func TestSortWriter(t *testing.T) {
+	f := func(maxBufLen, maxLines int, data string, expectedResult string) {
+		t.Helper()
+
+		var bb bytes.Buffer
+		sw := getSortWriter()
+		sw.Init(&bb, maxBufLen, maxLines)
+		for _, s := range strings.Split(data, "\n") {
+			if !sw.TryWrite([]byte(s + "\n")) {
+				break
+			}
+		}
+		sw.FinalFlush()
+		putSortWriter(sw)
+
+		result := bb.String()
+		if result != expectedResult {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, expectedResult)
+		}
+	}
+
+	f(100, 0, "", "")
+	f(100, 0, "{}", "{}\n")
+
+	data := `{"_time":"def","_msg":"xxx"}
+{"_time":"abc","_msg":"foo"}`
+	resultExpected := `{"_time":"abc","_msg":"foo"}
+{"_time":"def","_msg":"xxx"}
+`
+	f(100, 0, data, resultExpected)
+	f(10, 0, data, data+"\n")
+
+	// Test with the maxLines
+	f(100, 1, data, `{"_time":"abc","_msg":"foo"}`+"\n")
+	f(10, 1, data, `{"_time":"def","_msg":"xxx"}`+"\n")
+	f(10, 2, data, data+"\n")
+	f(100, 2, data, resultExpected)
+}

app/vlselect/main.go

@@ -0,0 +1,174 @@
+package vlselect
+
+import (
+	"context"
+	"embed"
+	"flag"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlselect/logsql"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+var (
+	maxConcurrentRequests = flag.Int("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "The maximum number of concurrent search requests. "+
+		"It shouldn't be high, since a single request can saturate all the CPU cores, while many concurrently executed requests may require high amounts of memory. "+
+		"See also -search.maxQueueDuration")
+	maxQueueDuration = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the search request waits for execution when -search.maxConcurrentRequests "+
+		"limit is reached; see also -search.maxQueryDuration")
+	maxQueryDuration = flag.Duration("search.maxQueryDuration", time.Second*30, "The maximum duration for query execution")
+)
+
+func getDefaultMaxConcurrentRequests() int {
+	n := cgroup.AvailableCPUs()
+	if n <= 4 {
+		n *= 2
+	}
+	if n > 16 {
+		// A single request can saturate all the CPU cores, so there is no sense
+		// in allowing higher number of concurrent requests - they will just contend
+		// for unavailable CPU time.
+		n = 16
+	}
+	return n
+}
+
+// Init initializes vlselect
+func Init() {
+	concurrencyLimitCh = make(chan struct{}, *maxConcurrentRequests)
+}
+
+// Stop stops vlselect
+func Stop() {
+}
+
+var concurrencyLimitCh chan struct{}
+
+var (
+	concurrencyLimitReached = metrics.NewCounter(`vl_concurrent_select_limit_reached_total`)
+	concurrencyLimitTimeout = metrics.NewCounter(`vl_concurrent_select_limit_timeout_total`)
+
+	_ = metrics.NewGauge(`vl_concurrent_select_capacity`, func() float64 {
+		return float64(cap(concurrencyLimitCh))
+	})
+	_ = metrics.NewGauge(`vl_concurrent_select_current`, func() float64 {
+		return float64(len(concurrencyLimitCh))
+	})
+)
+
+//go:embed vmui
+var vmuiFiles embed.FS
+
+var vmuiFileServer = http.FileServer(http.FS(vmuiFiles))
+
+// RequestHandler handles select requests for VictoriaLogs
+func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
+	path := r.URL.Path
+	if !strings.HasPrefix(path, "/select/") {
+		// Skip requests, which do not start with /select/, since these aren't our requests.
+		return false
+	}
+	path = strings.TrimPrefix(path, "/select")
+	path = strings.ReplaceAll(path, "//", "/")
+
+	if path == "/vmui" {
+		// VMUI access via incomplete url without `/` in the end. Redirect to complete url.
+		// Use relative redirect, since the hostname and path prefix may be incorrect if VictoriaMetrics
+		// is hidden behind vmauth or similar proxy.
+		_ = r.ParseForm()
+		newURL := "vmui/?" + r.Form.Encode()
+		httpserver.Redirect(w, newURL)
+		return true
+	}
+	if strings.HasPrefix(path, "/vmui/") {
+		if strings.HasPrefix(path, "/vmui/static/") {
+			// Allow clients caching static contents for long period of time, since it shouldn't change over time.
+			// Path to static contents (such as js and css) must be changed whenever its contents is changed.
+			// See https://developer.chrome.com/docs/lighthouse/performance/uses-long-cache-ttl/
+			w.Header().Set("Cache-Control", "max-age=31536000")
+		}
+		r.URL.Path = path
+		vmuiFileServer.ServeHTTP(w, r)
+		return true
+	}
+
+	// Limit the number of concurrent queries, which can consume big amounts of CPU.
+	startTime := time.Now()
+	ctx := r.Context()
+	stopCh := ctx.Done()
+	select {
+	case concurrencyLimitCh <- struct{}{}:
+		defer func() { <-concurrencyLimitCh }()
+	default:
+		// Sleep for a while until giving up. This should resolve short bursts in requests.
+		concurrencyLimitReached.Inc()
+		d := getMaxQueryDuration(r)
+		if d > *maxQueueDuration {
+			d = *maxQueueDuration
+		}
+		t := timerpool.Get(d)
+		select {
+		case concurrencyLimitCh <- struct{}{}:
+			timerpool.Put(t)
+			defer func() { <-concurrencyLimitCh }()
+		case <-stopCh:
+			timerpool.Put(t)
+			remoteAddr := httpserver.GetQuotedRemoteAddr(r)
+			requestURI := httpserver.GetRequestURI(r)
+			logger.Infof("client has cancelled the request after %.3f seconds: remoteAddr=%s, requestURI: %q",
+				time.Since(startTime).Seconds(), remoteAddr, requestURI)
+			return true
+		case <-t.C:
+			timerpool.Put(t)
+			concurrencyLimitTimeout.Inc()
+			err := &httpserver.ErrorWithStatusCode{
+				Err: fmt.Errorf("couldn't start executing the request in %.3f seconds, since -search.maxConcurrentRequests=%d concurrent requests "+
+					"are executed. Possible solutions: to reduce query load; to add more compute resources to the server; "+
+					"to increase -search.maxQueueDuration=%s; to increase -search.maxQueryDuration; to increase -search.maxConcurrentRequests",
+					d.Seconds(), *maxConcurrentRequests, maxQueueDuration),
+				StatusCode: http.StatusServiceUnavailable,
+			}
+			httpserver.Errorf(w, r, "%s", err)
+			return true
+		}
+	}
+
+	ctxWithCancel, cancel := context.WithCancel(ctx)
+	defer cancel()
+	stopCh = ctxWithCancel.Done()
+
+	switch {
+	case path == "/logsql/query":
+		logsqlQueryRequests.Inc()
+		httpserver.EnableCORS(w, r)
+		logsql.ProcessQueryRequest(w, r, stopCh, cancel)
+		return true
+	default:
+		return false
+	}
+}
+
+// getMaxQueryDuration returns the maximum duration for query from r.
+func getMaxQueryDuration(r *http.Request) time.Duration {
+	dms, err := httputils.GetDuration(r, "timeout", 0)
+	if err != nil {
+		dms = 0
+	}
+	d := time.Duration(dms) * time.Millisecond
+	if d <= 0 || d > *maxQueryDuration {
+		d = *maxQueryDuration
+	}
+	return d
+}
+
+var (
+	logsqlQueryRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/query"}`)
+)