mirror of
https://github.com/Icinga/icingadb.git
synced 2026-05-28 04:35:54 -04:00
Julian Brost
GitHub
commit
e0859c55a0
11 changed files with 114 additions and 354 deletions
2
.github/workflows/compliance.yml
vendored
2
.github/workflows/compliance.yml
vendored
|
|
@ -13,7 +13,7 @@ jobs:
|
|||
|
||||
- uses: actions/setup-go@v2
|
||||
with:
|
||||
go-version: 1.16.x
|
||||
go-version: 1.18.x
|
||||
|
||||
- uses: actions/checkout@v2
|
||||
|
||||
|
|
|
|||
10
.github/workflows/go.yml
vendored
10
.github/workflows/go.yml
vendored
|
|
@ -18,7 +18,7 @@ jobs:
|
|||
|
||||
- uses: actions/setup-go@v2
|
||||
with:
|
||||
go-version: 1.16.x
|
||||
go-version: 1.18.x
|
||||
|
||||
- run: go build -gcflags="-m" ./...
|
||||
|
||||
|
|
@ -31,7 +31,7 @@ jobs:
|
|||
|
||||
- uses: actions/setup-go@v2
|
||||
with:
|
||||
go-version: 1.16.x
|
||||
go-version: 1.18.x
|
||||
|
||||
- uses: dominikh/staticcheck-action@29e9b80fb8de0521ba4ed3fdf68fed5bbe82a2d2 # v1.1.0
|
||||
with:
|
||||
|
|
@ -44,7 +44,7 @@ jobs:
|
|||
|
||||
- uses: actions/setup-go@v2
|
||||
with:
|
||||
go-version: 1.16.x
|
||||
go-version: 1.18.x
|
||||
|
||||
- run: go vet ./...
|
||||
|
||||
|
|
@ -55,7 +55,7 @@ jobs:
|
|||
|
||||
- uses: actions/setup-go@v2
|
||||
with:
|
||||
go-version: 1.16.x
|
||||
go-version: 1.18.x
|
||||
|
||||
- name: Run gofmt -d .
|
||||
run: |
|
||||
|
|
@ -70,7 +70,7 @@ jobs:
|
|||
|
||||
- uses: actions/setup-go@v2
|
||||
with:
|
||||
go-version: 1.16.x
|
||||
go-version: 1.18.x
|
||||
|
||||
- name: Run go mod tidy
|
||||
run: |
|
||||
|
|
|
|||
2
.github/workflows/integration-tests.yml
vendored
2
.github/workflows/integration-tests.yml
vendored
|
|
@ -28,7 +28,7 @@ jobs:
|
|||
- name: Setup Go
|
||||
uses: actions/setup-go@v2
|
||||
with:
|
||||
go-version: '^1.16' # icinga-testing uses go:embed which is available in Go 1.16+
|
||||
go-version: '^1.18'
|
||||
- name: Build Icinga DB
|
||||
run: go build ./cmd/icingadb
|
||||
env:
|
||||
|
|
|
|||
18
go.mod
18
go.mod
|
|
@ -1,6 +1,6 @@
|
|||
module github.com/icinga/icingadb
|
||||
|
||||
go 1.16
|
||||
go 1.18
|
||||
|
||||
require (
|
||||
github.com/creasty/defaults v1.5.2
|
||||
|
|
@ -19,3 +19,19 @@ require (
|
|||
golang.org/x/exp v0.0.0-20210514180818-737f94c0881e
|
||||
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/benbjohnson/clock v1.1.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.1.2 // indirect
|
||||
github.com/davecgh/go-spew v1.1.1 // indirect
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
||||
github.com/fatih/color v1.10.0 // indirect
|
||||
github.com/mattn/go-colorable v0.1.8 // indirect
|
||||
github.com/mattn/go-isatty v0.0.12 // indirect
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
go.uber.org/atomic v1.7.0 // indirect
|
||||
go.uber.org/multierr v1.6.0 // indirect
|
||||
golang.org/x/sys v0.0.0-20210510120138-977fb7262007 // indirect
|
||||
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
|
||||
)
|
||||
|
|
|
|||
37
go.sum
37
go.sum
|
|
@ -1,11 +1,7 @@
|
|||
dmitri.shuralyov.com/gpu/mtl v0.0.0-20201218220906-28db891af037/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
|
||||
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
|
||||
github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
|
||||
github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
|
||||
github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
|
||||
github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
|
||||
github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
|
||||
github.com/creasty/defaults v1.5.2 h1:/VfB6uxpyp6h0fr7SPp7n8WJBoV8jfxQXPCnkVSjyls=
|
||||
github.com/creasty/defaults v1.5.2/go.mod h1:FPZ+Y0WNrbqOVw+c6av63eyHUAl6pMHZwqLPvXUZGfY=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
|
|
@ -18,7 +14,6 @@ github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGE
|
|||
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
|
||||
github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
|
||||
github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
|
||||
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
|
||||
github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
|
||||
github.com/go-playground/locales v0.13.0 h1:HyWk6mgj5qFqCT5fjGBuRArbVDfE4hi8+e8ceBS/t7Q=
|
||||
github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8=
|
||||
|
|
@ -48,7 +43,6 @@ github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
|
|||
github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
|
||||
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
|
||||
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
|
||||
|
|
@ -56,7 +50,6 @@ github.com/jessevdk/go-flags v1.5.0 h1:1jKYvbxEjfUl0fmqTCOfonvskHHXMjBySTLW4y9LF
|
|||
github.com/jessevdk/go-flags v1.5.0/go.mod h1:Fw0T6WPc1dYxT4mKEZRfG5kJhaTDP9pj1c2EWnYs/m4=
|
||||
github.com/jmoiron/sqlx v1.3.4 h1:wv+0IJZfL5z0uZoUjlpKgHkgaFSYD+r9CfrXjEXsO7w=
|
||||
github.com/jmoiron/sqlx v1.3.4/go.mod h1:2BljVx/86SuTyjE+aPYlHCTNvZrnJXghYGpNiXLBMCQ=
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
|
||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
|
|
@ -91,14 +84,9 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
|||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
|
||||
github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
|
||||
github.com/rs/zerolog v1.21.0/go.mod h1:ZPhntP/xmq1nnND05hhpAh2QMhSsA4UN3MGZ6O2J3hM=
|
||||
github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
|
||||
github.com/ssgreg/journald v1.0.0 h1:0YmTDPJXxcWDPba12qNMdO6TxvfkFSYpFIJ31CwmLcU=
|
||||
github.com/ssgreg/journald v1.0.0/go.mod h1:RUckwmTM8ghGWPslq2+ZBZzbb9/2KgjzYZ4JEP+oRt0=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
|
||||
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
|
||||
|
|
@ -106,35 +94,21 @@ github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5Cc
|
|||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
|
||||
go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
|
||||
go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
|
||||
go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
|
||||
go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI=
|
||||
go.uber.org/goleak v1.1.11/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
|
||||
go.uber.org/multierr v1.5.0/go.mod h1:FeouvMocqHpRaaGuG9EjoKcStLC43Zu/fmqdUMPcKYU=
|
||||
go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
|
||||
go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
|
||||
go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA=
|
||||
go.uber.org/zap v1.16.0/go.mod h1:MA8QOfq0BHJwdXa996Y4dYkAqRKB8/1K1QMMZVaNZjQ=
|
||||
go.uber.org/zap v1.21.0 h1:WefMeulhovoZ2sYXz7st6K0sLj7bBhpiFaud4r4zST8=
|
||||
go.uber.org/zap v1.21.0/go.mod h1:wjWOCqI0f2ZZrJF/UufIOkiC8ii6tm1iqIsLo76RfJw=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/exp v0.0.0-20190731235908-ec7cb31e5a56/go.mod h1:JhuoJpWY28nO4Vef9tZUw9qufEGTyX1+7lmHxV5q5G4=
|
||||
golang.org/x/exp v0.0.0-20210514180818-737f94c0881e h1:VqVU3dsTLGDa5pW74b+xG1lvKltt4EZIUrFPeKOqV2s=
|
||||
golang.org/x/exp v0.0.0-20210514180818-737f94c0881e/go.mod h1:MSdmUWF4ZWBPSUbgUX/gaau5kvnbkSs9pgtY6B9JXDE=
|
||||
golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
|
||||
golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
|
||||
golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
|
||||
golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE=
|
||||
golang.org/x/mobile v0.0.0-20201217150744-e6ae53a27f4f/go.mod h1:skQtrUTUwhdJvXM/2KKJzY8pDgNr9I/FOMqDVRPBUS4=
|
||||
golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
|
||||
golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY=
|
||||
golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
|
||||
golang.org/x/mod v0.1.1-0.20191209134235-331c550502dd/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
|
|
@ -156,7 +130,6 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
|
|||
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190904154756-749cb33beabd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
|
|
@ -164,7 +137,6 @@ golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7w
|
|||
golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210320140829-1e4c9ba3b0c4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
|
|
@ -178,14 +150,8 @@ golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
|
|||
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
|
||||
golang.org/x/tools v0.0.0-20190312151545-0bb0c0a6e846/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
|
||||
golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
|
||||
golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.0.0-20200117012304-6edc0a871e69/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
|
||||
golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
|
||||
golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
|
||||
golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
|
|
@ -203,7 +169,6 @@ google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ
|
|||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
|
||||
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
|
||||
gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
|
||||
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
|
||||
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
|
||||
|
|
@ -216,5 +181,3 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
|
|||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
|
||||
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
|
||||
honnef.co/go/tools v0.1.3/go.mod h1:NgwopIslSNH47DimFoV78dnkksY2EFtX0ajyb3K/las=
|
||||
|
|
|
|||
|
|
@ -2,40 +2,80 @@ package com
|
|||
|
||||
import (
|
||||
"context"
|
||||
"github.com/icinga/icingadb/pkg/contracts"
|
||||
"golang.org/x/sync/errgroup"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// BulkChunkSplitPolicy is a state machine which tracks the items of a chunk a bulker assembles.
|
||||
// A call takes an item for the current chunk into account.
|
||||
// Output true indicates that the state machine was reset first and the bulker
|
||||
// shall finish the current chunk now (not e.g. once $size is reached) without the given item.
|
||||
type BulkChunkSplitPolicy[T any] func(T) bool
|
||||
|
||||
type BulkChunkSplitPolicyFactory[T any] func() BulkChunkSplitPolicy[T]
|
||||
|
||||
// NeverSplit returns a pseudo state machine which never demands splitting.
|
||||
func NeverSplit[T any]() BulkChunkSplitPolicy[T] {
|
||||
return neverSplit[T]
|
||||
}
|
||||
|
||||
// SplitOnDupId returns a state machine which tracks the inputs' IDs.
|
||||
// Once an already seen input arrives, it demands splitting.
|
||||
func SplitOnDupId[T contracts.IDer]() BulkChunkSplitPolicy[T] {
|
||||
seenIds := map[string]struct{}{}
|
||||
|
||||
return func(ider T) bool {
|
||||
id := ider.ID().String()
|
||||
|
||||
_, ok := seenIds[id]
|
||||
if ok {
|
||||
seenIds = map[string]struct{}{id: {}}
|
||||
} else {
|
||||
seenIds[id] = struct{}{}
|
||||
}
|
||||
|
||||
return ok
|
||||
}
|
||||
}
|
||||
|
||||
func neverSplit[T any](T) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// Bulker reads all values from a channel and streams them in chunks into a Bulk channel.
|
||||
type Bulker struct {
|
||||
ch chan []interface{}
|
||||
type Bulker[T any] struct {
|
||||
ch chan []T
|
||||
ctx context.Context
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// NewBulker returns a new Bulker and starts streaming.
|
||||
func NewBulker(ctx context.Context, ch <-chan interface{}, count int) *Bulker {
|
||||
b := &Bulker{
|
||||
ch: make(chan []interface{}),
|
||||
func NewBulker[T any](
|
||||
ctx context.Context, ch <-chan T, count int, splitPolicyFactory BulkChunkSplitPolicyFactory[T],
|
||||
) *Bulker[T] {
|
||||
b := &Bulker[T]{
|
||||
ch: make(chan []T),
|
||||
ctx: ctx,
|
||||
mu: sync.Mutex{},
|
||||
}
|
||||
|
||||
go b.run(ch, count)
|
||||
go b.run(ch, count, splitPolicyFactory)
|
||||
|
||||
return b
|
||||
}
|
||||
|
||||
// Bulk returns the channel on which the bulks are delivered.
|
||||
func (b *Bulker) Bulk() <-chan []interface{} {
|
||||
func (b *Bulker[T]) Bulk() <-chan []T {
|
||||
return b.ch
|
||||
}
|
||||
|
||||
func (b *Bulker) run(ch <-chan interface{}, count int) {
|
||||
func (b *Bulker[T]) run(ch <-chan T, count int, splitPolicyFactory BulkChunkSplitPolicyFactory[T]) {
|
||||
defer close(b.ch)
|
||||
|
||||
bufCh := make(chan interface{}, count)
|
||||
bufCh := make(chan T, count)
|
||||
splitPolicy := splitPolicyFactory()
|
||||
g, ctx := errgroup.WithContext(b.ctx)
|
||||
|
||||
g.Go(func() error {
|
||||
|
|
@ -57,7 +97,7 @@ func (b *Bulker) run(ch <-chan interface{}, count int) {
|
|||
|
||||
g.Go(func() error {
|
||||
for done := false; !done; {
|
||||
buf := make([]interface{}, 0, count)
|
||||
buf := make([]T, 0, count)
|
||||
timeout := time.After(256 * time.Millisecond)
|
||||
|
||||
for drain := true; drain && len(buf) < count; {
|
||||
|
|
@ -70,6 +110,15 @@ func (b *Bulker) run(ch <-chan interface{}, count int) {
|
|||
break
|
||||
}
|
||||
|
||||
if splitPolicy(v) {
|
||||
if len(buf) > 0 {
|
||||
b.ch <- buf
|
||||
buf = make([]T, 0, count)
|
||||
}
|
||||
|
||||
timeout = time.After(256 * time.Millisecond)
|
||||
}
|
||||
|
||||
buf = append(buf, v)
|
||||
case <-timeout:
|
||||
drain = false
|
||||
|
|
@ -81,6 +130,8 @@ func (b *Bulker) run(ch <-chan interface{}, count int) {
|
|||
if len(buf) > 0 {
|
||||
b.ch <- buf
|
||||
}
|
||||
|
||||
splitPolicy = splitPolicyFactory()
|
||||
}
|
||||
|
||||
return nil
|
||||
|
|
@ -92,18 +143,20 @@ func (b *Bulker) run(ch <-chan interface{}, count int) {
|
|||
}
|
||||
|
||||
// Bulk reads all values from a channel and streams them in chunks into a returned channel.
|
||||
func Bulk(ctx context.Context, ch <-chan interface{}, count int) <-chan []interface{} {
|
||||
func Bulk[T any](
|
||||
ctx context.Context, ch <-chan T, count int, splitPolicyFactory BulkChunkSplitPolicyFactory[T],
|
||||
) <-chan []T {
|
||||
if count <= 1 {
|
||||
return oneBulk(ctx, ch)
|
||||
}
|
||||
|
||||
return NewBulker(ctx, ch, count).Bulk()
|
||||
return NewBulker(ctx, ch, count, splitPolicyFactory).Bulk()
|
||||
}
|
||||
|
||||
// oneBulk operates just as NewBulker(ctx, ch, 1).Bulk(),
|
||||
// but without the overhead of the actual bulk creation with a buffer channel and timeout.
|
||||
func oneBulk(ctx context.Context, ch <-chan interface{}) <-chan []interface{} {
|
||||
out := make(chan []interface{})
|
||||
// oneBulk operates just as NewBulker(ctx, ch, 1, splitPolicy).Bulk(),
|
||||
// but without the overhead of the actual bulk creation with a buffer channel, timeout and BulkChunkSplitPolicy.
|
||||
func oneBulk[T any](ctx context.Context, ch <-chan T) <-chan []T {
|
||||
out := make(chan []T)
|
||||
go func() {
|
||||
defer close(out)
|
||||
|
||||
|
|
@ -111,7 +164,7 @@ func oneBulk(ctx context.Context, ch <-chan interface{}) <-chan []interface{} {
|
|||
select {
|
||||
case item := <-ch:
|
||||
select {
|
||||
case out <- []interface{}{item}:
|
||||
case out <- []T{item}:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
|
@ -123,3 +176,8 @@ func oneBulk(ctx context.Context, ch <-chan interface{}) <-chan []interface{} {
|
|||
|
||||
return out
|
||||
}
|
||||
|
||||
var (
|
||||
_ BulkChunkSplitPolicyFactory[struct{}] = NeverSplit[struct{}]
|
||||
_ BulkChunkSplitPolicyFactory[contracts.Entity] = SplitOnDupId[contracts.Entity]
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,183 +0,0 @@
|
|||
package com
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/icinga/icingadb/pkg/contracts"
|
||||
"golang.org/x/sync/errgroup"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// BulkChunkSplitPolicy is a state machine which tracks the items of a chunk a bulker assembles.
|
||||
// A call takes an item for the current chunk into account.
|
||||
// Output true indicates that the state machine was reset first and the bulker
|
||||
// shall finish the current chunk now (not e.g. once $size is reached) without the given item.
|
||||
type BulkChunkSplitPolicy func(contracts.Entity) bool
|
||||
|
||||
type BulkChunkSplitPolicyFactory func() BulkChunkSplitPolicy
|
||||
|
||||
// NeverSplit returns a pseudo state machine which never demands splitting.
|
||||
func NeverSplit() BulkChunkSplitPolicy {
|
||||
return neverSplit
|
||||
}
|
||||
|
||||
// SplitOnDupId returns a state machine which tracks the inputs' IDs.
|
||||
// Once an already seen input arrives, it demands splitting.
|
||||
func SplitOnDupId() BulkChunkSplitPolicy {
|
||||
seenIds := map[string]struct{}{}
|
||||
|
||||
return func(entity contracts.Entity) bool {
|
||||
id := entity.ID().String()
|
||||
|
||||
_, ok := seenIds[id]
|
||||
if ok {
|
||||
seenIds = map[string]struct{}{id: {}}
|
||||
} else {
|
||||
seenIds[id] = struct{}{}
|
||||
}
|
||||
|
||||
return ok
|
||||
}
|
||||
}
|
||||
|
||||
func neverSplit(contracts.Entity) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// EntityBulker reads all entities from a channel and streams them in chunks into a Bulk channel.
|
||||
type EntityBulker struct {
|
||||
ch chan []contracts.Entity
|
||||
ctx context.Context
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// NewEntityBulker returns a new EntityBulker and starts streaming.
|
||||
func NewEntityBulker(
|
||||
ctx context.Context, ch <-chan contracts.Entity, count int, splitPolicyFactory BulkChunkSplitPolicyFactory,
|
||||
) *EntityBulker {
|
||||
b := &EntityBulker{
|
||||
ch: make(chan []contracts.Entity),
|
||||
ctx: ctx,
|
||||
mu: sync.Mutex{},
|
||||
}
|
||||
|
||||
go b.run(ch, count, splitPolicyFactory)
|
||||
|
||||
return b
|
||||
}
|
||||
|
||||
// Bulk returns the channel on which the bulks are delivered.
|
||||
func (b *EntityBulker) Bulk() <-chan []contracts.Entity {
|
||||
return b.ch
|
||||
}
|
||||
|
||||
func (b *EntityBulker) run(ch <-chan contracts.Entity, count int, splitPolicyFactory BulkChunkSplitPolicyFactory) {
|
||||
defer close(b.ch)
|
||||
|
||||
bufCh := make(chan contracts.Entity, count)
|
||||
splitPolicy := splitPolicyFactory()
|
||||
g, ctx := errgroup.WithContext(b.ctx)
|
||||
|
||||
g.Go(func() error {
|
||||
defer close(bufCh)
|
||||
|
||||
for {
|
||||
select {
|
||||
case v, ok := <-ch:
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
bufCh <- v
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
g.Go(func() error {
|
||||
for done := false; !done; {
|
||||
buf := make([]contracts.Entity, 0, count)
|
||||
timeout := time.After(256 * time.Millisecond)
|
||||
|
||||
for drain := true; drain && len(buf) < count; {
|
||||
select {
|
||||
case v, ok := <-bufCh:
|
||||
if !ok {
|
||||
drain = false
|
||||
done = true
|
||||
|
||||
break
|
||||
}
|
||||
|
||||
if splitPolicy(v) {
|
||||
if len(buf) > 0 {
|
||||
b.ch <- buf
|
||||
buf = make([]contracts.Entity, 0, count)
|
||||
}
|
||||
|
||||
timeout = time.After(256 * time.Millisecond)
|
||||
}
|
||||
|
||||
buf = append(buf, v)
|
||||
case <-timeout:
|
||||
drain = false
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
if len(buf) > 0 {
|
||||
b.ch <- buf
|
||||
}
|
||||
|
||||
splitPolicy = splitPolicyFactory()
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
// We don't expect an error here.
|
||||
// We only use errgroup for the encapsulated use of sync.WaitGroup.
|
||||
_ = g.Wait()
|
||||
}
|
||||
|
||||
// BulkEntities reads all entities from a channel and streams them in chunks into a returned channel.
|
||||
func BulkEntities(
|
||||
ctx context.Context, ch <-chan contracts.Entity, count int, splitPolicyFactory BulkChunkSplitPolicyFactory,
|
||||
) <-chan []contracts.Entity {
|
||||
if count <= 1 {
|
||||
return oneEntityBulk(ctx, ch)
|
||||
}
|
||||
|
||||
return NewEntityBulker(ctx, ch, count, splitPolicyFactory).Bulk()
|
||||
}
|
||||
|
||||
// oneEntityBulk operates just as NewEntityBulker(ctx, ch, 1, splitPolicy).Bulk(),
|
||||
// but without the overhead of the actual bulk creation with a buffer channel, timeout and BulkChunkSplitPolicy.
|
||||
func oneEntityBulk(ctx context.Context, ch <-chan contracts.Entity) <-chan []contracts.Entity {
|
||||
out := make(chan []contracts.Entity)
|
||||
go func() {
|
||||
defer close(out)
|
||||
|
||||
for {
|
||||
select {
|
||||
case item := <-ch:
|
||||
select {
|
||||
case out <- []contracts.Entity{item}:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return out
|
||||
}
|
||||
|
||||
var (
|
||||
_ BulkChunkSplitPolicyFactory = NeverSplit
|
||||
_ BulkChunkSplitPolicyFactory = SplitOnDupId
|
||||
)
|
||||
|
|
@ -1,98 +0,0 @@
|
|||
package com
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/go-redis/redis/v8"
|
||||
"golang.org/x/sync/errgroup"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// XMessageBulker reads all values from a channel and streams them in chunks into a Bulk channel.
|
||||
type XMessageBulker struct {
|
||||
ch chan []redis.XMessage
|
||||
ctx context.Context
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// NewXMessageBulker returns a new XMessageBulker and starts streaming.
|
||||
func NewXMessageBulker(ctx context.Context, ch <-chan redis.XMessage, count int) *XMessageBulker {
|
||||
b := &XMessageBulker{
|
||||
ch: make(chan []redis.XMessage),
|
||||
ctx: ctx,
|
||||
mu: sync.Mutex{},
|
||||
}
|
||||
|
||||
go b.run(ch, count)
|
||||
|
||||
return b
|
||||
}
|
||||
|
||||
// Bulk returns the channel on which the bulks are delivered.
|
||||
func (b *XMessageBulker) Bulk() <-chan []redis.XMessage {
|
||||
return b.ch
|
||||
}
|
||||
|
||||
func (b *XMessageBulker) run(ch <-chan redis.XMessage, count int) {
|
||||
defer close(b.ch)
|
||||
|
||||
bufCh := make(chan redis.XMessage, count)
|
||||
g, ctx := errgroup.WithContext(b.ctx)
|
||||
|
||||
g.Go(func() error {
|
||||
defer close(bufCh)
|
||||
|
||||
for {
|
||||
select {
|
||||
case v, ok := <-ch:
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
bufCh <- v
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
g.Go(func() error {
|
||||
for done := false; !done; {
|
||||
buf := make([]redis.XMessage, 0, count)
|
||||
timeout := time.After(256 * time.Millisecond)
|
||||
|
||||
for drain := true; drain && len(buf) < count; {
|
||||
select {
|
||||
case v, ok := <-bufCh:
|
||||
if !ok {
|
||||
drain = false
|
||||
done = true
|
||||
|
||||
break
|
||||
}
|
||||
|
||||
buf = append(buf, v)
|
||||
case <-timeout:
|
||||
drain = false
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
if len(buf) > 0 {
|
||||
b.ch <- buf
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
// We don't expect an error here.
|
||||
// We only use errgroup for the encapsulated use of sync.WaitGroup.
|
||||
_ = g.Wait()
|
||||
}
|
||||
|
||||
// BulkXMessages reads all values from a channel and streams them in chunks into a returned channel.
|
||||
func BulkXMessages(ctx context.Context, ch <-chan redis.XMessage, count int) <-chan []redis.XMessage {
|
||||
return NewXMessageBulker(ctx, ch, count).Bulk()
|
||||
}
|
||||
|
|
@ -240,7 +240,7 @@ func (db *DB) BulkExec(ctx context.Context, query string, count int, sem *semaph
|
|||
|
||||
g, ctx := errgroup.WithContext(ctx)
|
||||
// Use context from group.
|
||||
bulk := com.Bulk(ctx, arg, count)
|
||||
bulk := com.Bulk(ctx, arg, count, com.NeverSplit[any])
|
||||
|
||||
g.Go(func() error {
|
||||
g, ctx := errgroup.WithContext(ctx)
|
||||
|
|
@ -304,14 +304,14 @@ func (db *DB) BulkExec(ctx context.Context, query string, count int, sem *semaph
|
|||
// and can be executed concurrently to the extent allowed by the semaphore passed in sem.
|
||||
// Entities for which the query ran successfully will be streamed on the succeeded channel.
|
||||
func (db *DB) NamedBulkExec(
|
||||
ctx context.Context, query string, count int, sem *semaphore.Weighted,
|
||||
arg <-chan contracts.Entity, succeeded chan<- contracts.Entity, splitPolicyFactory com.BulkChunkSplitPolicyFactory,
|
||||
ctx context.Context, query string, count int, sem *semaphore.Weighted, arg <-chan contracts.Entity,
|
||||
succeeded chan<- contracts.Entity, splitPolicyFactory com.BulkChunkSplitPolicyFactory[contracts.Entity],
|
||||
) error {
|
||||
var counter com.Counter
|
||||
defer db.log(ctx, query, &counter).Stop()
|
||||
|
||||
g, ctx := errgroup.WithContext(ctx)
|
||||
bulk := com.BulkEntities(ctx, arg, count, splitPolicyFactory)
|
||||
bulk := com.Bulk(ctx, arg, count, splitPolicyFactory)
|
||||
|
||||
g.Go(func() error {
|
||||
for {
|
||||
|
|
@ -379,7 +379,7 @@ func (db *DB) NamedBulkExecTx(
|
|||
defer db.log(ctx, query, &counter).Stop()
|
||||
|
||||
g, ctx := errgroup.WithContext(ctx)
|
||||
bulk := com.BulkEntities(ctx, arg, count, com.NeverSplit)
|
||||
bulk := com.Bulk(ctx, arg, count, com.NeverSplit[contracts.Entity])
|
||||
|
||||
g.Go(func() error {
|
||||
for {
|
||||
|
|
@ -502,7 +502,9 @@ func (db *DB) CreateStreamed(ctx context.Context, entities <-chan contracts.Enti
|
|||
sem := db.GetSemaphoreForTable(utils.TableName(first))
|
||||
stmt, placeholders := db.BuildInsertStmt(first)
|
||||
|
||||
return db.NamedBulkExec(ctx, stmt, db.BatchSizeByPlaceholders(placeholders), sem, forward, nil, com.NeverSplit)
|
||||
return db.NamedBulkExec(
|
||||
ctx, stmt, db.BatchSizeByPlaceholders(placeholders), sem, forward, nil, com.NeverSplit[contracts.Entity],
|
||||
)
|
||||
}
|
||||
|
||||
// UpsertStreamed bulk upserts the specified entities via NamedBulkExec.
|
||||
|
|
@ -519,7 +521,8 @@ func (db *DB) UpsertStreamed(ctx context.Context, entities <-chan contracts.Enti
|
|||
stmt, placeholders := db.BuildUpsertStmt(first)
|
||||
|
||||
return db.NamedBulkExec(
|
||||
ctx, stmt, db.BatchSizeByPlaceholders(placeholders), sem, forward, succeeded, com.SplitOnDupId,
|
||||
ctx, stmt, db.BatchSizeByPlaceholders(placeholders), sem,
|
||||
forward, succeeded, com.SplitOnDupId[contracts.Entity],
|
||||
)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -141,7 +141,7 @@ func (s Sync) deleteFromRedis(ctx context.Context, key string, input <-chan redi
|
|||
}
|
||||
}).Stop()
|
||||
|
||||
bulks := com.BulkXMessages(ctx, input, s.redis.Options.HScanCount)
|
||||
bulks := com.Bulk(ctx, input, s.redis.Options.HScanCount, com.NeverSplit[redis.XMessage])
|
||||
stream := "icinga:history:stream:" + key
|
||||
for {
|
||||
select {
|
||||
|
|
|
|||
|
|
@ -109,7 +109,7 @@ func (r *RuntimeUpdates) Sync(
|
|||
sem := semaphore.NewWeighted(1)
|
||||
|
||||
return r.db.NamedBulkExec(
|
||||
ctx, upsertStmt, upsertCount, sem, upsertEntities, upserted, com.SplitOnDupId,
|
||||
ctx, upsertStmt, upsertCount, sem, upsertEntities, upserted, com.SplitOnDupId[contracts.Entity],
|
||||
)
|
||||
})
|
||||
g.Go(func() error {
|
||||
|
|
@ -213,7 +213,7 @@ func (r *RuntimeUpdates) Sync(
|
|||
sem := semaphore.NewWeighted(1)
|
||||
|
||||
return r.db.NamedBulkExec(
|
||||
ctx, cvStmt, cvCount, sem, customvars, upsertedCustomvars, com.SplitOnDupId,
|
||||
ctx, cvStmt, cvCount, sem, customvars, upsertedCustomvars, com.SplitOnDupId[contracts.Entity],
|
||||
)
|
||||
})
|
||||
g.Go(func() error {
|
||||
|
|
@ -248,7 +248,8 @@ func (r *RuntimeUpdates) Sync(
|
|||
sem := semaphore.NewWeighted(1)
|
||||
|
||||
return r.db.NamedBulkExec(
|
||||
ctx, cvFlatStmt, cvFlatCount, sem, flatCustomvars, upsertedFlatCustomvars, com.SplitOnDupId,
|
||||
ctx, cvFlatStmt, cvFlatCount, sem, flatCustomvars,
|
||||
upsertedFlatCustomvars, com.SplitOnDupId[contracts.Entity],
|
||||
)
|
||||
})
|
||||
g.Go(func() error {
|
||||
|
|
|
|||
Loading…
Reference in a new issue