mirror of
https://github.com/caiwx86/small-packages.git
synced 2026-11-02 01:03:56 +08:00
action committed
1 parent
ccd677de3a
commit
825070ba95
2037 files changed
+1694705
-12285
No files matched your search
+28
-63
@@ -5,23 +5,17 @@
|
||||
include $(TOPDIR)/rules.mk
|
||||
|
||||
PKG_NAME:=daed
|
||||
PKG_VERSION:=2026.08.26
|
||||
DAED_VERSION:=daed-671e65d
|
||||
WING_VERSION:=wing-dc50308
|
||||
CORE_VERSION:=core-5db27a0
|
||||
WING_HASH_SHORT:=$(shell echo $(WING_VERSION) | cut -d- -f2)
|
||||
CORE_HASH_SHORT:=$(shell echo $(CORE_VERSION) | cut -d- -f2)
|
||||
PKG_RELEASE:=1
|
||||
PKG_VERSION:=2026.08.28
|
||||
PKG_RELEASE:=4
|
||||
|
||||
PKG_SOURCE:=$(PKG_NAME)-$(PKG_VERSION).tar.gz
|
||||
PKG_SOURCE_PROTO:=git
|
||||
PKG_SOURCE_VERSION:=671e65d2fdcd62fe6a3ec18ecda209c5addea898
|
||||
PKG_SOURCE_URL:=https://github.com/daeuniverse/daed.git
|
||||
PKG_MIRROR_HASH:=skip
|
||||
PKG_SOURCE:=daed-src-2026.08.28-73f01b995c34.tar.gz
|
||||
PKG_SOURCE_URL:=https://github.com/kenzok8/openwrt-daede/releases/download/daed-src
|
||||
PKG_SOURCE_SUBDIR:=$(PKG_NAME)-$(PKG_VERSION)
|
||||
PKG_HASH:=73f01b995c341586fdf7ab6ee8ed7a4cb52b59fd913ca75b1e61b3ecb6c9a350
|
||||
|
||||
PKG_LICENSE:=AGPL-3.0-only MIT
|
||||
PKG_LICENSE_FILES:=LICENSE wing/LICENSE
|
||||
PKG_MAINTAINER:=Tianling Shen <cnsztl@immortalwrt.org>
|
||||
PKG_MAINTAINER:=kenzok8
|
||||
|
||||
DAED_BUILD_DIR:=$(BUILD_DIR)/$(PKG_NAME)-$(PKG_VERSION)
|
||||
PKG_BUILD_DIR:=$(DAED_BUILD_DIR)/wing
|
||||
@@ -30,6 +24,13 @@ PKG_BUILD_DEPENDS:=golang/host bpf-headers
|
||||
PKG_BUILD_PARALLEL:=1
|
||||
PKG_BUILD_FLAGS:=no-mips16
|
||||
|
||||
# armv7 lacks a CO-RE vmlinux.h for the trace eBPF (under wing/dae-core), so
|
||||
# trace's bpf2go fails there. Apply sbwml/openwrt_helloworld's arm patches
|
||||
# (add vmlinux-arm.h + drop the kprobe that doesn't work) to build on arm.
|
||||
ifeq ($(ARCH),arm)
|
||||
PATCH_DIR:=$(CURDIR)/patches_arm
|
||||
endif
|
||||
|
||||
GO_PKG:=github.com/daeuniverse/dae-wing
|
||||
|
||||
GO_PKG_LDFLAGS:= \
|
||||
@@ -38,16 +39,14 @@ GO_PKG_LDFLAGS:= \
|
||||
-X '$(GO_PKG)/db.AppDescription=$(PKG_NAME) is a integration solution of dae, API and UI.'
|
||||
GO_PKG_LDFLAGS_X:= \
|
||||
$(GO_PKG)/db.AppName=$(PKG_NAME) \
|
||||
$(GO_PKG)/db.AppVersion=$(DAED_VERSION)_$(WING_VERSION)_$(CORE_VERSION)
|
||||
GO_PKG_TAGS:=embedallowed,trace
|
||||
$(GO_PKG)/db.AppVersion=$(PKG_VERSION)
|
||||
GO_PKG_TAGS:=embedallowed,trace,timetzdata
|
||||
|
||||
include $(INCLUDE_DIR)/package.mk
|
||||
include $(INCLUDE_DIR)/bpf.mk
|
||||
include $(TOPDIR)/feeds/packages/lang/golang/golang-package.mk
|
||||
|
||||
GO_PKG_BUILD_VARS+= \
|
||||
GOFLAGS="-trimpath -buildvcs=false -pgo=auto"
|
||||
|
||||
GO_PKG_BUILD_VARS+= GOFLAGS="-trimpath -buildvcs=false -pgo=auto"
|
||||
GO_PKG_TARGET_VARS+= \
|
||||
CGO_LDFLAGS="$(TARGET_LDFLAGS) -static -Wl,-s" \
|
||||
GOEXPERIMENT=newinliner,simd
|
||||
@@ -94,52 +93,11 @@ define Package/daed/conffiles
|
||||
/etc/config/daed
|
||||
endef
|
||||
|
||||
NODE_VERSION:=v24.12.0
|
||||
NODE_DIST:=node-$(NODE_VERSION)-linux-x64
|
||||
NODE_URL:=https://nodejs.org/dist/$(NODE_VERSION)/$(NODE_DIST).tar.xz
|
||||
|
||||
define Build/Prepare
|
||||
( \
|
||||
rm -rf $(DAED_BUILD_DIR) ; \
|
||||
mkdir -p $(DAED_BUILD_DIR) ; \
|
||||
$(TAR) --strip-components=1 -C $(DAED_BUILD_DIR) -xzf $(DL_DIR)/$(PKG_NAME)-$(PKG_VERSION).tar.gz ; \
|
||||
git clone https://github.com/daeuniverse/dae-wing $(PKG_BUILD_DIR) && \
|
||||
git -C $(PKG_BUILD_DIR) checkout $(WING_HASH_SHORT) ; \
|
||||
rm -rf $(PKG_BUILD_DIR)/dae-core ; \
|
||||
git clone https://github.com/daeuniverse/dae $(PKG_BUILD_DIR)/dae-core && \
|
||||
git -C $(PKG_BUILD_DIR)/dae-core checkout $(CORE_HASH_SHORT) ; \
|
||||
pushd $(PKG_BUILD_DIR)/dae-core ; \
|
||||
git submodule update --init ; \
|
||||
go get -u=patch ; \
|
||||
go mod tidy ; \
|
||||
popd ; \
|
||||
pushd $(PKG_BUILD_DIR) ; \
|
||||
go get -u=patch ; \
|
||||
go mod tidy ; \
|
||||
wget -qO default.pgo "https://github.com/QiuSimons/luci-app-dae/raw/refs/heads/kix/dae/pprof/default.pgo" ; \
|
||||
popd ; \
|
||||
mkdir -p $(DAED_BUILD_DIR)/.node_tmp/config ; \
|
||||
wget -qO - "$(NODE_URL)" | tar -xJ -C $(DAED_BUILD_DIR)/.node_tmp --strip-components=1 ; \
|
||||
export PATH="$(DAED_BUILD_DIR)/.node_tmp/bin:$$$$PATH" ; \
|
||||
export XDG_CONFIG_HOME="$(DAED_BUILD_DIR)/.node_tmp/config" ; \
|
||||
npm install -g pnpm ; \
|
||||
pushd $(DAED_BUILD_DIR) ; \
|
||||
pnpm install ; \
|
||||
pnpm build --filter daed ; \
|
||||
popd ; \
|
||||
mkdir -p $(PKG_BUILD_DIR)/webrender/web ; \
|
||||
cp -rf $(DAED_BUILD_DIR)/apps/web/dist/* $(PKG_BUILD_DIR)/webrender/web ; \
|
||||
find $(PKG_BUILD_DIR)/webrender/web -name "*.map" -type f -delete ; \
|
||||
find $(PKG_BUILD_DIR)/webrender/web -type f -size +4k ! -name "*.gz" ! -name "*.woff" ! -name "*.woff2" -exec sh -c '\
|
||||
gzip -9 -k "{}"; \
|
||||
if [ "$$$$(stat -c %s {})" -lt "$$$$(stat -c %s {}.gz)" ]; then \
|
||||
rm {}.gz; \
|
||||
else \
|
||||
rm {}; \
|
||||
fi' \
|
||||
";" ; \
|
||||
rm -rf $(DAED_BUILD_DIR)/.node_tmp ; \
|
||||
)
|
||||
mkdir -p $(DAED_BUILD_DIR)
|
||||
$(TAR) --strip-components=1 -C $(DAED_BUILD_DIR) -xzf $(DL_DIR)/$(PKG_SOURCE)
|
||||
$(if $(wildcard $(CURDIR)/patches/*.patch),$(call PatchDir,$(PKG_BUILD_DIR),$(CURDIR)/patches,))
|
||||
$(if $(filter arm,$(ARCH)),$(if $(wildcard $(CURDIR)/patches_arm/*.patch),$(call PatchDir,$(PKG_BUILD_DIR),$(CURDIR)/patches_arm,)))
|
||||
endef
|
||||
|
||||
DAE_CFLAGS:= \
|
||||
@@ -174,12 +132,19 @@ define Package/daed/install
|
||||
$(call GoPackage/Package/Install/Bin,$(PKG_INSTALL_DIR))
|
||||
$(INSTALL_DIR) $(1)/usr/bin
|
||||
$(INSTALL_BIN) $(PKG_INSTALL_DIR)/usr/bin/dae-wing $(1)/usr/bin/daed
|
||||
$(INSTALL_BIN) $(CURDIR)/files/daed-guard $(1)/usr/bin/daed-guard
|
||||
|
||||
$(INSTALL_DIR) $(1)/usr/share/daed
|
||||
$(INSTALL_DATA) $(CURDIR)/files/daed-cleanup.sh $(1)/usr/share/daed/cleanup.sh
|
||||
|
||||
$(INSTALL_DIR) $(1)/etc/config
|
||||
$(INSTALL_CONF) $(CURDIR)/files/daed.config $(1)/etc/config/daed
|
||||
|
||||
$(INSTALL_DIR) $(1)/etc/init.d
|
||||
$(INSTALL_BIN) $(CURDIR)/files/daed.init $(1)/etc/init.d/daed
|
||||
|
||||
$(INSTALL_DIR) $(1)/lib/upgrade/keep.d
|
||||
$(INSTALL_DATA) $(CURDIR)/files/daed.keep $(1)/lib/upgrade/keep.d/daed
|
||||
endef
|
||||
|
||||
$(eval $(call GoBinPackage,daed))
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
#!/bin/sh
|
||||
|
||||
daed_cleanup_runtime() {
|
||||
local pid
|
||||
|
||||
for pid in $(ip netns pids daens 2>/dev/null); do
|
||||
kill "$pid" 2>/dev/null
|
||||
done
|
||||
|
||||
if [ -n "$(ip netns pids daens 2>/dev/null)" ]; then
|
||||
sleep 1
|
||||
for pid in $(ip netns pids daens 2>/dev/null); do
|
||||
kill -9 "$pid" 2>/dev/null
|
||||
done
|
||||
fi
|
||||
|
||||
if ! ip netns del daens 2>/dev/null; then
|
||||
umount -l /run/netns/daens 2>/dev/null
|
||||
rm -f /run/netns/daens
|
||||
fi
|
||||
|
||||
ip link del dae0 2>/dev/null || true
|
||||
|
||||
[ ! -e /run/netns/daens ] || return 1
|
||||
! ip netns list 2>/dev/null | awk '$1 == "daens" { found=1 } END { exit !found }' || return 1
|
||||
! ip link show dae0 >/dev/null 2>&1 || return 1
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
#!/bin/sh
|
||||
|
||||
. /usr/share/daed/cleanup.sh
|
||||
|
||||
if ! daed_cleanup_runtime; then
|
||||
echo "daed: stale network state could not be removed" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Keep daed as a child so a panic or unexpected exit is followed by an
|
||||
# immediate teardown of all kernel/runtime state before procd can respawn us.
|
||||
child_pid=
|
||||
cleanup_child() {
|
||||
[ -n "$child_pid" ] && kill "$child_pid" 2>/dev/null
|
||||
}
|
||||
trap cleanup_child TERM INT
|
||||
|
||||
/usr/bin/daed "$@" &
|
||||
child_pid=$!
|
||||
wait "$child_pid"
|
||||
status=$?
|
||||
child_pid=
|
||||
trap - TERM INT
|
||||
|
||||
if ! daed_cleanup_runtime; then
|
||||
echo "daed: runtime cleanup after exit failed" >&2
|
||||
status=1
|
||||
fi
|
||||
exit "$status"
|
||||
@@ -4,4 +4,6 @@ config daed 'config'
|
||||
option listen_addr '0.0.0.0:2023'
|
||||
option log_maxbackups '1'
|
||||
option log_maxsize '5'
|
||||
|
||||
option subscribe_auto_update '0'
|
||||
option subscribe_update_cycle 'daily'
|
||||
option subscribe_update_hour '4'
|
||||
Executable → Regular
+17
-14
@@ -1,20 +1,14 @@
|
||||
#!/bin/sh /etc/rc.common
|
||||
# Copyright (C) 2023 Tianling Shen <cnsztl@immortalwrt.org>
|
||||
|
||||
USE_PROCD=1
|
||||
START=99
|
||||
|
||||
CONF="daed"
|
||||
PROG="/usr/bin/daed"
|
||||
PROG="/usr/bin/daed-guard"
|
||||
LOG="/var/log/daed/daed.log"
|
||||
|
||||
|
||||
cleanup_netns() {
|
||||
if [ -e "/run/netns/daens" ]; then
|
||||
logger -t daed "Cleaning up residual netns 'daens'..."
|
||||
ip netns del daens 2>/dev/null
|
||||
rm -f /run/netns/daens 2>/dev/null
|
||||
fi
|
||||
}
|
||||
. /usr/share/daed/cleanup.sh
|
||||
|
||||
start_service() {
|
||||
config_load "$CONF"
|
||||
@@ -23,14 +17,13 @@ start_service() {
|
||||
config_get_bool enabled "config" "enabled" "0"
|
||||
[ "$enabled" -eq "1" ] || return 1
|
||||
|
||||
cleanup_netns
|
||||
|
||||
local listen_addr log_maxbackups log_maxsize
|
||||
config_get listen_addr "config" "listen_addr" "0.0.0.0:2023"
|
||||
config_get log_maxbackups "config" "log_maxbackups" "1"
|
||||
config_get log_maxsize "config" "log_maxsize" "5"
|
||||
|
||||
procd_open_instance "$CONF"
|
||||
procd_set_param env DAE_LOCATION_ASSET="/usr/share/v2ray" TZ="$(uci -q get system.@system[0].zonename)"
|
||||
procd_set_param command "$PROG" run
|
||||
procd_append_param command --config "/etc/daed/"
|
||||
procd_append_param command --listen "$listen_addr"
|
||||
@@ -40,15 +33,25 @@ start_service() {
|
||||
|
||||
procd_set_param limits core="unlimited"
|
||||
procd_set_param limits nofile="1000000 1000000"
|
||||
procd_set_param respawn
|
||||
# Avoid an endless crash/respawn loop which can repeatedly reattach dae's
|
||||
# data-plane hooks and make the router management plane unreachable.
|
||||
procd_set_param respawn 3600 5 5
|
||||
# procd_set_param stdout 1
|
||||
procd_set_param stderr 1
|
||||
|
||||
procd_close_instance
|
||||
}
|
||||
|
||||
stop_service() {
|
||||
procd_kill "$CONF" ""
|
||||
cleanup_netns
|
||||
rm -f "$LOG"
|
||||
daed_cleanup_runtime
|
||||
}
|
||||
|
||||
restart() {
|
||||
stop
|
||||
sleep 1
|
||||
daed_cleanup_runtime
|
||||
start
|
||||
}
|
||||
|
||||
service_triggers() {
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
/etc/daed/wing.db-wal
|
||||
@@ -0,0 +1,23 @@
|
||||
From: kenzok8 <kenzok8@noreply>
|
||||
Date: 2026-06-23
|
||||
Subject: [PATCH] daed: detach BPF hooks when stopping from dashboard
|
||||
|
||||
Detach BPF hooks when the dashboard stops dae through EmptyConfig.
|
||||
Reported: https://github.com/kenzok8/openwrt-daede/issues/13
|
||||
---
|
||||
--- a/dae/run.go
|
||||
+++ b/dae/run.go
|
||||
@@ -138,6 +138,13 @@
|
||||
/* dae-wing start */
|
||||
newConf := newReloadMsg.Config
|
||||
/* dae-wing end */
|
||||
+
|
||||
+ // openwrt-daede: dashboard stop reloads to EmptyConfig; detach hooks.
|
||||
+ if newConf == EmptyConfig {
|
||||
+ if e := c.DetachBpfHooks(); e != nil {
|
||||
+ log.Errorln("[Stop] DetachBpfHooks:", e)
|
||||
+ }
|
||||
+ }
|
||||
// New logger.
|
||||
oldLogOutput := log.Out
|
||||
log = logrus.New()
|
||||
@@ -0,0 +1,18 @@
|
||||
From: kenzok8 <kenzok8@noreply>
|
||||
Date: 2026-06-29
|
||||
Subject: [PATCH] daed: use bounded probes for manual latency tests
|
||||
|
||||
Use the existing bounded probe for manual dashboard latency tests.
|
||||
Reported: https://github.com/kenzok8/openwrt-daede/issues/20
|
||||
---
|
||||
--- a/graphql/service/node/latency_mutation_utils.go
|
||||
+++ b/graphql/service/node/latency_mutation_utils.go
|
||||
@@ -108,7 +108,7 @@ func testSingleNodeLatency(option *diale
|
||||
}
|
||||
defer d.Close()
|
||||
|
||||
- result, err := d.ProbeLatency()
|
||||
+ result, err := d.ProbeLatencyFast()
|
||||
if err != nil {
|
||||
msg := err.Error()
|
||||
resolver.MessageV = &msg
|
||||
@@ -0,0 +1,19 @@
|
||||
From: kenzok8 <kenzok8@noreply>
|
||||
Date: 2026-06-30
|
||||
Subject: [PATCH] daed: raise manual latency probe concurrency to 16
|
||||
|
||||
Raise concurrent probes 8->16 so manual latency tests finish faster
|
||||
when many nodes (incl. dead/timeout ones) are present.
|
||||
Reported: https://github.com/kenzok8/openwrt-daede/issues/20
|
||||
---
|
||||
--- a/graphql/service/node/latency_mutation_utils.go
|
||||
+++ b/graphql/service/node/latency_mutation_utils.go
|
||||
@@ -19,7 +19,7 @@
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
-const latencyProbeConcurrency = 8
|
||||
+const latencyProbeConcurrency = 16
|
||||
|
||||
func TestLatencies(ctx context.Context, ids *[]graphql.ID) ([]*LatencyResolver, error) {
|
||||
option, err := latencyProbeOption(ctx)
|
||||
@@ -0,0 +1,63 @@
|
||||
diff --git a/graphql/service/config/mutation_utils.go b/graphql/service/config/mutation_utils.go
|
||||
index 7a3c33f..fd61554 100644
|
||||
--- a/graphql/service/config/mutation_utils.go
|
||||
+++ b/graphql/service/config/mutation_utils.go
|
||||
@@ -9,11 +9,13 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
+ "os"
|
||||
"reflect"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
+ "time"
|
||||
|
||||
"github.com/daeuniverse/dae-wing/common"
|
||||
"github.com/daeuniverse/dae-wing/dae"
|
||||
@@ -258,6 +260,15 @@ func Rename(ctx context.Context, _id graphql.ID, name string) (n int32, err erro
|
||||
|
||||
var runLock sync.Mutex
|
||||
|
||||
+const reloadWaitTimeout = 75 * time.Second
|
||||
+
|
||||
+func restartAfterReloadTimeout() {
|
||||
+ go func() {
|
||||
+ time.Sleep(time.Second)
|
||||
+ os.Exit(1)
|
||||
+ }()
|
||||
+}
|
||||
+
|
||||
func Run(d *gorm.DB, noLoad bool) (n int32, err error) {
|
||||
if ok := runLock.TryLock(); !ok {
|
||||
return 0, fmt.Errorf("the last request didn't complete; make a cup of tea and take a break")
|
||||
@@ -455,13 +466,24 @@ func Run(d *gorm.DB, noLoad bool) (n int32, err error) {
|
||||
|
||||
/// Reload with current config.
|
||||
chReloadCallback := make(chan error)
|
||||
- dae.ChReloadConfigs <- &dae.ReloadMessage{
|
||||
+ reloadMsg := &dae.ReloadMessage{
|
||||
Config: c,
|
||||
Callback: chReloadCallback,
|
||||
}
|
||||
- errReload := <-chReloadCallback
|
||||
- if errReload != nil {
|
||||
- return 0, fmt.Errorf("failed to load new config: %w; see more in log", errReload)
|
||||
+ select {
|
||||
+ case dae.ChReloadConfigs <- reloadMsg:
|
||||
+ case <-time.After(reloadWaitTimeout):
|
||||
+ restartAfterReloadTimeout()
|
||||
+ return 0, fmt.Errorf("failed to submit reload within %s; restarting daed to recover runtime", reloadWaitTimeout)
|
||||
+ }
|
||||
+ select {
|
||||
+ case errReload := <-chReloadCallback:
|
||||
+ if errReload != nil {
|
||||
+ return 0, fmt.Errorf("failed to load new config: %w; see more in log", errReload)
|
||||
+ }
|
||||
+ case <-time.After(reloadWaitTimeout):
|
||||
+ restartAfterReloadTimeout()
|
||||
+ return 0, fmt.Errorf("reload timed out after %s; restarting daed to recover runtime", reloadWaitTimeout)
|
||||
}
|
||||
|
||||
// Save running status
|
||||
@@ -0,0 +1,88 @@
|
||||
diff --git a/dae/run.go b/dae/run.go
|
||||
index d8f86c2..a0a665d 100644
|
||||
--- a/dae/run.go
|
||||
+++ b/dae/run.go
|
||||
@@ -32,7 +32,8 @@ var ChReloadConfigs = make(chan *ReloadMessage)
|
||||
var GracefullyExit = make(chan struct{})
|
||||
var EmptyConfig *daeConfig.Config
|
||||
var c *control.ControlPlane
|
||||
-var onceWaitingNetwork sync.Once
|
||||
+var waitingNetworkMu sync.Mutex
|
||||
+var networkReady bool
|
||||
|
||||
func init() {
|
||||
sections, err := config_parser.Parse(`global{} routing{}`)
|
||||
@@ -52,6 +53,19 @@ func ControlPlane() (*control.ControlPlane, error) {
|
||||
return c, nil
|
||||
}
|
||||
|
||||
+func waitForNetwork(log *logrus.Logger) error {
|
||||
+ waitingNetworkMu.Lock()
|
||||
+ defer waitingNetworkMu.Unlock()
|
||||
+ if networkReady {
|
||||
+ return nil
|
||||
+ }
|
||||
+ if err := WaitForNetwork(log); err != nil {
|
||||
+ return err
|
||||
+ }
|
||||
+ networkReady = true
|
||||
+ return nil
|
||||
+}
|
||||
+
|
||||
func Run(log *logrus.Logger, conf *daeConfig.Config, externGeoDataDirs []string, disableTimestamp bool, dry bool) (err error) {
|
||||
defer close(GracefullyExit)
|
||||
// Not really run dae.
|
||||
@@ -221,6 +235,9 @@ func newControlPlane(log *logrus.Logger, bpf interface{}, dnsCache map[string]*c
|
||||
|
||||
// Deep copy to prevent modification.
|
||||
conf = deepcopy.Copy(conf).(*daeConfig.Config)
|
||||
+ if bpf == nil {
|
||||
+ control.PurgeStaleTCFilters(log)
|
||||
+ }
|
||||
|
||||
// Init Direct Dialers.
|
||||
direct.InitDirectDialers(conf.Global.FallbackResolver)
|
||||
@@ -228,9 +245,9 @@ func newControlPlane(log *logrus.Logger, bpf interface{}, dnsCache map[string]*c
|
||||
|
||||
if !conf.Global.DisableWaitingNetwork && len(conf.Global.WanInterface) > 0 {
|
||||
// Wait for network for WAN ready.
|
||||
- onceWaitingNetwork.Do(func() {
|
||||
- WaitForNetwork(log)
|
||||
- })
|
||||
+ if err = waitForNetwork(log); err != nil {
|
||||
+ return nil, err
|
||||
+ }
|
||||
}
|
||||
|
||||
/// Get subscription -> nodeList mapping.
|
||||
diff --git a/dae/utils.go b/dae/utils.go
|
||||
index 65965a2..dc43cbc 100644
|
||||
--- a/dae/utils.go
|
||||
+++ b/dae/utils.go
|
||||
@@ -94,8 +94,9 @@ func preprocessWanInterfaceAuto(params *daeConfig.Config) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
-func WaitForNetwork(log *logrus.Logger) {
|
||||
+func WaitForNetwork(log *logrus.Logger) error {
|
||||
epo := 5 * time.Second
|
||||
+ deadline := time.Now().Add(60 * time.Second)
|
||||
client := http.Client{
|
||||
Transport: &http.Transport{
|
||||
DialContext: func(ctx context.Context, network, addr string) (c net.Conn, err error) {
|
||||
@@ -115,6 +116,9 @@ func WaitForNetwork(log *logrus.Logger) {
|
||||
}
|
||||
log.Infoln("Waiting for network...")
|
||||
for i := 0; ; i++ {
|
||||
+ if time.Now().After(deadline) {
|
||||
+ return fmt.Errorf("network unavailable after 60 seconds")
|
||||
+ }
|
||||
resp, err := client.Get(cmd.CheckNetworkLinks[i%len(cmd.CheckNetworkLinks)])
|
||||
if err != nil {
|
||||
log.Debugln("CheckNetwork:", err)
|
||||
@@ -134,4 +138,5 @@ func WaitForNetwork(log *logrus.Logger) {
|
||||
time.Sleep(epo)
|
||||
}
|
||||
log.Infoln("Network online.")
|
||||
+ return nil
|
||||
}
|
||||
@@ -0,0 +1,617 @@
|
||||
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
|
||||
Subject: [PATCH] daed: strictly reconcile subscription updates
|
||||
MIME-Version: 1.0
|
||||
Content-Type: text/plain; charset=UTF-8
|
||||
Content-Transfer-Encoding: 8bit
|
||||
|
||||
订阅更新按节点链接精确对账，普通组里的失效节点删除，fixed 组固定节点转成独立节点保留。
|
||||
|
||||
数据库提交后按运行状态应用配置，补齐并发地址校验、导入失败回滚和 reload 超时恢复。
|
||||
---
|
||||
cmd/run.go | 6 +-
|
||||
graphql/mutation.go | 9 +-
|
||||
graphql/root_schema.go | 2 +-
|
||||
graphql/service/config/mutation_utils.go | 59 ++++-
|
||||
.../service/subscription/mutation_utils.go | 181 +++++++++++----
|
||||
.../subscription/mutation_utils_test.go | 212 ++++++++++++++++++
|
||||
6 files changed, 408 insertions(+), 61 deletions(-)
|
||||
create mode 100644 graphql/service/subscription/mutation_utils_test.go
|
||||
|
||||
diff --git a/cmd/run.go b/cmd/run.go
|
||||
index 7dce577..978f46c 100644
|
||||
--- a/cmd/run.go
|
||||
+++ b/cmd/run.go
|
||||
@@ -161,11 +161,8 @@ func restoreRunningState() (err error) {
|
||||
if !reload {
|
||||
return nil
|
||||
}
|
||||
- tx := db.BeginTx(context.TODO())
|
||||
// Reload.
|
||||
- if _, err = config.Run(tx, false); err != nil {
|
||||
- tx.Rollback()
|
||||
-
|
||||
+ if _, err = config.Run(context.TODO(), false); err != nil {
|
||||
// Another tx.
|
||||
// Set running = false.
|
||||
tx2 := db.BeginTx(context.TODO())
|
||||
@@ -183,7 +180,6 @@ func restoreRunningState() (err error) {
|
||||
tx2.Commit()
|
||||
return err
|
||||
}
|
||||
- tx.Commit()
|
||||
return nil
|
||||
}
|
||||
|
||||
diff --git a/graphql/mutation.go b/graphql/mutation.go
|
||||
index fcbf440..adb747c 100644
|
||||
--- a/graphql/mutation.go
|
||||
+++ b/graphql/mutation.go
|
||||
@@ -256,14 +256,7 @@ func (r *MutationResolver) SelectConfig(args *struct {
|
||||
func (r *MutationResolver) Run(args *struct {
|
||||
Dry bool
|
||||
}) (int32, error) {
|
||||
- tx := db.BeginTx(context.TODO())
|
||||
- ret, err := config.Run(tx, args.Dry)
|
||||
- if err != nil {
|
||||
- tx.Rollback()
|
||||
- return 0, err
|
||||
- }
|
||||
- tx.Commit()
|
||||
- return ret, nil
|
||||
+ return config.Run(context.TODO(), args.Dry)
|
||||
}
|
||||
|
||||
func (r *MutationResolver) CreateDns(args *struct {
|
||||
diff --git a/graphql/root_schema.go b/graphql/root_schema.go
|
||||
index 8b82db6..a96cace 100644
|
||||
--- a/graphql/root_schema.go
|
||||
+++ b/graphql/root_schema.go
|
||||
@@ -121,7 +121,7 @@ type Mutation {
|
||||
# tagSubscription is to give the subscription a new tag.
|
||||
tagSubscription(id: ID!, tag: String!): Int! @hasRole(role: ADMIN)
|
||||
|
||||
- # updateSubscription is to re-fetch subscription and resolve subscription into nodes. Old nodes that independently belong to any groups will not be removed.
|
||||
+ # updateSubscription re-fetches and strictly reconciles subscription nodes. Stale nodes pinned by fixed groups become independent nodes.
|
||||
updateSubscription(id: ID!): Subscription! @hasRole(role: ADMIN)
|
||||
|
||||
# updateSubscriptionLink is to update the subscription link without re-fetching nodes.
|
||||
diff --git a/graphql/service/config/mutation_utils.go b/graphql/service/config/mutation_utils.go
|
||||
index fd61554..feb94d3 100644
|
||||
--- a/graphql/service/config/mutation_utils.go
|
||||
+++ b/graphql/service/config/mutation_utils.go
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
"github.com/daeuniverse/dae-wing/dae"
|
||||
"github.com/daeuniverse/dae-wing/db"
|
||||
"github.com/daeuniverse/dae-wing/graphql/service/config/global"
|
||||
+ "github.com/daeuniverse/dae-wing/graphql/service/general"
|
||||
daeConfig "github.com/daeuniverse/dae/config"
|
||||
"github.com/daeuniverse/dae/pkg/config_parser"
|
||||
"github.com/graph-gophers/graphql-go"
|
||||
@@ -269,21 +270,69 @@ func restartAfterReloadTimeout() {
|
||||
}()
|
||||
}
|
||||
|
||||
-func Run(d *gorm.DB, noLoad bool) (n int32, err error) {
|
||||
+func runTransaction(ctx context.Context, noLoad bool) (n int32, err error) {
|
||||
+ tx := db.BeginTx(ctx)
|
||||
+ if tx.Error != nil {
|
||||
+ return 0, tx.Error
|
||||
+ }
|
||||
+ n, err = runLocked(tx, noLoad)
|
||||
+ if err != nil {
|
||||
+ tx.Rollback()
|
||||
+ return 0, err
|
||||
+ }
|
||||
+ if err = tx.Commit().Error; err != nil {
|
||||
+ return 0, err
|
||||
+ }
|
||||
+ return n, nil
|
||||
+}
|
||||
+
|
||||
+func Run(ctx context.Context, noLoad bool) (n int32, err error) {
|
||||
if ok := runLock.TryLock(); !ok {
|
||||
return 0, fmt.Errorf("the last request didn't complete; make a cup of tea and take a break")
|
||||
}
|
||||
defer runLock.Unlock()
|
||||
+ return runTransaction(ctx, noLoad)
|
||||
+}
|
||||
+
|
||||
+func ApplyIfRunning(ctx context.Context) error {
|
||||
+ runLock.Lock()
|
||||
+ defer runLock.Unlock()
|
||||
+
|
||||
+ daeResolver := general.DaeResolver{Ctx: ctx}
|
||||
+ running, err := daeResolver.Running()
|
||||
+ if err != nil || !running {
|
||||
+ return err
|
||||
+ }
|
||||
+ modified, err := daeResolver.Modified()
|
||||
+ if err != nil || !modified {
|
||||
+ return err
|
||||
+ }
|
||||
+ _, err = runTransaction(ctx, false)
|
||||
+ return err
|
||||
+}
|
||||
+
|
||||
+func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
|
||||
//// Dry run.
|
||||
if noLoad {
|
||||
ch := make(chan error)
|
||||
- dae.ChReloadConfigs <- &dae.ReloadMessage{
|
||||
+ reloadMsg := &dae.ReloadMessage{
|
||||
Config: dae.EmptyConfig,
|
||||
Callback: ch,
|
||||
}
|
||||
- err = <-ch
|
||||
- if err != nil {
|
||||
- return 0, fmt.Errorf("failed to dryrun: %w; see more in log and report bugs", err)
|
||||
+ select {
|
||||
+ case dae.ChReloadConfigs <- reloadMsg:
|
||||
+ case <-time.After(reloadWaitTimeout):
|
||||
+ restartAfterReloadTimeout()
|
||||
+ return 0, fmt.Errorf("failed to submit dryrun within %s; restarting daed to recover runtime", reloadWaitTimeout)
|
||||
+ }
|
||||
+ select {
|
||||
+ case err = <-ch:
|
||||
+ if err != nil {
|
||||
+ return 0, fmt.Errorf("failed to dryrun: %w; see more in log and report bugs", err)
|
||||
+ }
|
||||
+ case <-time.After(reloadWaitTimeout):
|
||||
+ restartAfterReloadTimeout()
|
||||
+ return 0, fmt.Errorf("dryrun timed out after %s; restarting daed to recover runtime", reloadWaitTimeout)
|
||||
}
|
||||
|
||||
// Running -> false
|
||||
diff --git a/graphql/service/subscription/mutation_utils.go b/graphql/service/subscription/mutation_utils.go
|
||||
index e55e65b..e01a342 100644
|
||||
--- a/graphql/service/subscription/mutation_utils.go
|
||||
+++ b/graphql/service/subscription/mutation_utils.go
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
"github.com/daeuniverse/dae-wing/dae"
|
||||
"github.com/daeuniverse/dae-wing/db"
|
||||
"github.com/daeuniverse/dae-wing/graphql/internal"
|
||||
+ "github.com/daeuniverse/dae-wing/graphql/service/config"
|
||||
"github.com/daeuniverse/dae-wing/graphql/service/node"
|
||||
"github.com/daeuniverse/dae/common/subscription"
|
||||
"github.com/go-co-op/gocron"
|
||||
@@ -171,8 +172,9 @@ func AutoUpdateVersionByIds(d *gorm.DB, ids []uint) (err error) {
|
||||
}
|
||||
|
||||
var (
|
||||
- schedulerCache = make(map[uint]*gocron.Scheduler)
|
||||
- schedulerMu sync.RWMutex
|
||||
+ schedulerCache = make(map[uint]*gocron.Scheduler)
|
||||
+ schedulerMu sync.RWMutex
|
||||
+ subscriptionUpdateMu sync.Mutex
|
||||
)
|
||||
|
||||
func UpdateAll(ctx context.Context) {
|
||||
@@ -258,68 +260,161 @@ func Update(ctx context.Context, _id graphql.ID) (r *Resolver, err error) {
|
||||
return &Resolver{Subscription: m}, nil
|
||||
}
|
||||
|
||||
-func UpdateById(ctx context.Context, subId uint) (sub *db.Subscription, err error) {
|
||||
- // Fetch node links.
|
||||
- var m db.Subscription
|
||||
- if err = db.DB(ctx).Where(&db.Subscription{ID: subId}).First(&m).Error; err != nil {
|
||||
- return nil, err
|
||||
+func reconcileSubscriptionNodes(tx *gorm.DB, subId uint, links []string) error {
|
||||
+ var existing []db.Node
|
||||
+ if err := tx.Where("subscription_id = ?", subId).Find(&existing).Error; err != nil {
|
||||
+ return err
|
||||
}
|
||||
- links, err := fetchLinks(m.Link)
|
||||
- if err != nil {
|
||||
- return nil, err
|
||||
+
|
||||
+ incoming := make(map[string]struct{}, len(links))
|
||||
+ for _, link := range links {
|
||||
+ incoming[link] = struct{}{}
|
||||
}
|
||||
|
||||
- tx := db.BeginTx(ctx)
|
||||
- defer func() {
|
||||
- if err == nil {
|
||||
- tx.Commit()
|
||||
+ current := make(map[string]struct{}, len(existing))
|
||||
+ var staleIds []uint
|
||||
+ for _, n := range existing {
|
||||
+ if _, ok := incoming[n.Link]; ok {
|
||||
+ current[n.Link] = struct{}{}
|
||||
+ continue
|
||||
+ }
|
||||
+ staleIds = append(staleIds, n.ID)
|
||||
+ }
|
||||
+
|
||||
+ fixed := make(map[uint]struct{})
|
||||
+ if len(staleIds) > 0 {
|
||||
+ var fixedIds []uint
|
||||
+ if err := tx.Table("group_nodes").
|
||||
+ Distinct("group_nodes.node_id").
|
||||
+ Joins("INNER JOIN groups ON groups.id = group_nodes.group_id").
|
||||
+ Where("groups.policy = ?", "fixed").
|
||||
+ Where("group_nodes.node_id IN ?", staleIds).
|
||||
+ Pluck("group_nodes.node_id", &fixedIds).Error; err != nil {
|
||||
+ return err
|
||||
+ }
|
||||
+ for _, id := range fixedIds {
|
||||
+ fixed[id] = struct{}{}
|
||||
+ }
|
||||
+ }
|
||||
+
|
||||
+ var detachIds, deleteIds []uint
|
||||
+ for _, id := range staleIds {
|
||||
+ if _, ok := fixed[id]; ok {
|
||||
+ detachIds = append(detachIds, id)
|
||||
} else {
|
||||
- tx.Rollback()
|
||||
+ deleteIds = append(deleteIds, id)
|
||||
+ }
|
||||
+ }
|
||||
+
|
||||
+ if len(staleIds) > 0 {
|
||||
+ if err := node.AutoUpdateVersionByIds(tx, staleIds); err != nil {
|
||||
+ return err
|
||||
+ }
|
||||
+ }
|
||||
+ if len(detachIds) > 0 {
|
||||
+ if err := tx.Model(&db.Node{}).
|
||||
+ Where("id IN ?", detachIds).
|
||||
+ Update("subscription_id", nil).Error; err != nil {
|
||||
+ return err
|
||||
+ }
|
||||
+ }
|
||||
+ if len(deleteIds) > 0 {
|
||||
+ if err := tx.Exec("DELETE FROM group_nodes WHERE node_id IN ?", deleteIds).Error; err != nil {
|
||||
+ return err
|
||||
+ }
|
||||
+ if err := tx.Where("id IN ?", deleteIds).Delete(&db.Node{}).Error; err != nil {
|
||||
+ return err
|
||||
}
|
||||
- }()
|
||||
- // Remove those nodes whose subscription are independent from any groups.
|
||||
- subQuery := tx.Raw(`select nodes.id as id
|
||||
- from nodes
|
||||
- inner join group_nodes on group_nodes.node_id = nodes.id
|
||||
- where subscription_id = ?`, subId)
|
||||
-
|
||||
- if err = tx.Where("subscription_id = ?", subId).
|
||||
- Where("id not in (?)", subQuery).
|
||||
- Select(clause.Associations).
|
||||
- Delete(&db.Node{}).Error; err != nil {
|
||||
- return nil, err
|
||||
}
|
||||
- // Import node links.
|
||||
+
|
||||
+ seen := make(map[string]struct{}, len(links))
|
||||
var args []*internal.ImportArgument
|
||||
for _, link := range links {
|
||||
+ if _, ok := seen[link]; ok {
|
||||
+ continue
|
||||
+ }
|
||||
+ seen[link] = struct{}{}
|
||||
+ if _, ok := current[link]; ok {
|
||||
+ continue
|
||||
+ }
|
||||
args = append(args, &internal.ImportArgument{Link: link})
|
||||
}
|
||||
- result, err := node.Import(tx, false, &subId, args)
|
||||
+ result, err := node.Import(tx, true, &subId, args)
|
||||
if err != nil {
|
||||
- return nil, err
|
||||
+ return errors.New("failed to import subscription node")
|
||||
}
|
||||
- hasAnyCandidate := false
|
||||
for _, r := range result {
|
||||
- if r.Error == nil {
|
||||
- hasAnyCandidate = true
|
||||
- break
|
||||
+ if r.Error != nil {
|
||||
+ return errors.New("failed to import subscription node")
|
||||
}
|
||||
}
|
||||
- if !hasAnyCandidate {
|
||||
- return nil, fmt.Errorf("interrupt to update subscription: no any valid node can be imported")
|
||||
+
|
||||
+ var count int64
|
||||
+ if err := tx.Model(&db.Node{}).Where("subscription_id = ?", subId).Count(&count).Error; err != nil {
|
||||
+ return err
|
||||
}
|
||||
- // Update updated_at and return the latest version.
|
||||
- if err = tx.Model(&m).
|
||||
+ if count == 0 {
|
||||
+ return fmt.Errorf("interrupt to update subscription: no any valid node can be imported")
|
||||
+ }
|
||||
+ return nil
|
||||
+}
|
||||
+
|
||||
+func updateSubscriptionTx(ctx context.Context, m *db.Subscription, subId uint, links []string) (err error) {
|
||||
+ subscriptionUpdateMu.Lock()
|
||||
+ defer subscriptionUpdateMu.Unlock()
|
||||
+
|
||||
+ tx := db.BeginTx(ctx)
|
||||
+ if tx.Error != nil {
|
||||
+ return tx.Error
|
||||
+ }
|
||||
+ defer func() {
|
||||
+ if err != nil {
|
||||
+ tx.Rollback()
|
||||
+ }
|
||||
+ }()
|
||||
+ var current db.Subscription
|
||||
+ if err = tx.Where(&db.Subscription{ID: subId}).First(¤t).Error; err != nil {
|
||||
+ return err
|
||||
+ }
|
||||
+ if current.Link != m.Link {
|
||||
+ return fmt.Errorf("subscription changed during update; retry")
|
||||
+ }
|
||||
+ *m = current
|
||||
+
|
||||
+ if err = reconcileSubscriptionNodes(tx, subId, links); err != nil {
|
||||
+ return err
|
||||
+ }
|
||||
+ q := tx.Model(m).
|
||||
Clauses(clause.Returning{}).
|
||||
Where(&db.Subscription{ID: subId}).
|
||||
- Update("updated_at", time.Now()).Error; err != nil {
|
||||
- return nil, err
|
||||
+ Update("updated_at", time.Now())
|
||||
+ if q.Error != nil {
|
||||
+ return q.Error
|
||||
+ }
|
||||
+ if q.RowsAffected == 0 {
|
||||
+ return fmt.Errorf("no such subscription")
|
||||
}
|
||||
-
|
||||
- // Update modified if subscription is referenced by running config.
|
||||
if err = AutoUpdateVersionByIds(tx, []uint{subId}); err != nil {
|
||||
+ return err
|
||||
+ }
|
||||
+ return tx.Commit().Error
|
||||
+}
|
||||
+
|
||||
+func UpdateById(ctx context.Context, subId uint) (sub *db.Subscription, err error) {
|
||||
+ var m db.Subscription
|
||||
+ if err = db.DB(ctx).Where(&db.Subscription{ID: subId}).First(&m).Error; err != nil {
|
||||
+ return nil, err
|
||||
+ }
|
||||
+ links, err := fetchLinks(m.Link)
|
||||
+ if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
+ if err = updateSubscriptionTx(ctx, &m, subId, links); err != nil {
|
||||
+ return nil, err
|
||||
+ }
|
||||
+ if err = config.ApplyIfRunning(ctx); err != nil {
|
||||
+ return nil, fmt.Errorf("subscription updated but failed to apply runtime: %w", err)
|
||||
+ }
|
||||
return &m, nil
|
||||
}
|
||||
|
||||
@@ -399,6 +494,8 @@ func UpdateLink(ctx context.Context, _id graphql.ID, link string) (r *Resolver,
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
+ subscriptionUpdateMu.Lock()
|
||||
+ defer subscriptionUpdateMu.Unlock()
|
||||
|
||||
tx := db.BeginTx(ctx)
|
||||
defer func() {
|
||||
diff --git a/graphql/service/subscription/mutation_utils_test.go b/graphql/service/subscription/mutation_utils_test.go
|
||||
new file mode 100644
|
||||
index 0000000..18ff9b9
|
||||
--- /dev/null
|
||||
+++ b/graphql/service/subscription/mutation_utils_test.go
|
||||
@@ -0,0 +1,212 @@
|
||||
+/*
|
||||
+ * SPDX-License-Identifier: AGPL-3.0-only
|
||||
+ * Copyright (c) 2023, daeuniverse Organization <team@v2raya.org>
|
||||
+ */
|
||||
+
|
||||
+package subscription
|
||||
+
|
||||
+import (
|
||||
+ "context"
|
||||
+ "fmt"
|
||||
+ "testing"
|
||||
+ "time"
|
||||
+
|
||||
+ "github.com/daeuniverse/dae-wing/db"
|
||||
+)
|
||||
+
|
||||
+func mustNoError(t *testing.T, err error) {
|
||||
+ t.Helper()
|
||||
+ if err != nil {
|
||||
+ t.Fatal(err)
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
+func assertError(t *testing.T, err error) {
|
||||
+ t.Helper()
|
||||
+ if err == nil {
|
||||
+ t.Fatal("expected error")
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
+func assertEqualValues(t *testing.T, want, got interface{}) {
|
||||
+ t.Helper()
|
||||
+ if fmt.Sprint(want) != fmt.Sprint(got) {
|
||||
+ t.Fatalf("want %v, got %v", want, got)
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
+func assertNotNil(t *testing.T, value *uint) {
|
||||
+ t.Helper()
|
||||
+ if value == nil {
|
||||
+ t.Fatal("expected non-nil value")
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
+func assertNil(t *testing.T, value *uint) {
|
||||
+ t.Helper()
|
||||
+ if value != nil {
|
||||
+ t.Fatalf("expected nil, got %v", *value)
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
+func assertZero(t *testing.T, value int64) {
|
||||
+ t.Helper()
|
||||
+ if value != 0 {
|
||||
+ t.Fatalf("expected zero, got %d", value)
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
+func initReconcileTestDB(t *testing.T) context.Context {
|
||||
+ t.Helper()
|
||||
+ mustNoError(t, db.InitDatabase(t.TempDir()))
|
||||
+ return context.Background()
|
||||
+}
|
||||
+
|
||||
+func createTestSubscription(t *testing.T, ctx context.Context, links ...string) (db.Subscription, []db.Node) {
|
||||
+ t.Helper()
|
||||
+ sub := db.Subscription{
|
||||
+ UpdatedAt: time.Now(),
|
||||
+ Link: "https://example.invalid/subscription",
|
||||
+ Status: "",
|
||||
+ Info: "",
|
||||
+ }
|
||||
+ mustNoError(t, db.DB(ctx).Create(&sub).Error)
|
||||
+
|
||||
+ nodes := make([]db.Node, 0, len(links))
|
||||
+ for i, link := range links {
|
||||
+ node := db.Node{
|
||||
+ Link: link,
|
||||
+ Name: "node",
|
||||
+ Address: "192.0.2.1",
|
||||
+ Protocol: "ss",
|
||||
+ SubscriptionID: &sub.ID,
|
||||
+ }
|
||||
+ node.Name += string(rune('A' + i))
|
||||
+ mustNoError(t, db.DB(ctx).Create(&node).Error)
|
||||
+ nodes = append(nodes, node)
|
||||
+ }
|
||||
+ return sub, nodes
|
||||
+}
|
||||
+
|
||||
+func createTestGroup(t *testing.T, ctx context.Context, policy string, systemID *uint, nodes ...db.Node) db.Group {
|
||||
+ t.Helper()
|
||||
+ group := db.Group{Name: policy + time.Now().Format("150405.000000000"), Policy: policy, SystemID: systemID}
|
||||
+ mustNoError(t, db.DB(ctx).Create(&group).Error)
|
||||
+ mustNoError(t, db.DB(ctx).Model(&group).Association("Node").Append(nodes))
|
||||
+ return group
|
||||
+}
|
||||
+
|
||||
+func countGroupNode(t *testing.T, ctx context.Context, groupID, nodeID uint) int64 {
|
||||
+ t.Helper()
|
||||
+ var count int64
|
||||
+ mustNoError(t, db.DB(ctx).Table("group_nodes").
|
||||
+ Where("group_id = ? AND node_id = ?", groupID, nodeID).
|
||||
+ Count(&count).Error)
|
||||
+ return count
|
||||
+}
|
||||
+
|
||||
+func TestReconcileKeepsCurrentReferencedNode(t *testing.T) {
|
||||
+ ctx := initReconcileTestDB(t)
|
||||
+ sub, nodes := createTestSubscription(t, ctx, "ss://current")
|
||||
+ group := createTestGroup(t, ctx, "min_moving_avg", nil, nodes[0])
|
||||
+
|
||||
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://current"}))
|
||||
+
|
||||
+ var got db.Node
|
||||
+ mustNoError(t, db.DB(ctx).First(&got, nodes[0].ID).Error)
|
||||
+ assertNotNil(t, got.SubscriptionID)
|
||||
+ assertEqualValues(t, sub.ID, *got.SubscriptionID)
|
||||
+ assertEqualValues(t, 1, countGroupNode(t, ctx, group.ID, got.ID))
|
||||
+}
|
||||
+
|
||||
+func TestReconcileDeletesStaleUnreferencedNode(t *testing.T) {
|
||||
+ ctx := initReconcileTestDB(t)
|
||||
+ sub, nodes := createTestSubscription(t, ctx, "ss://current", "ss://stale")
|
||||
+
|
||||
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://current"}))
|
||||
+
|
||||
+ var count int64
|
||||
+ mustNoError(t, db.DB(ctx).Model(&db.Node{}).Where("id = ?", nodes[1].ID).Count(&count).Error)
|
||||
+ assertZero(t, count)
|
||||
+}
|
||||
+
|
||||
+func TestReconcileDeletesStaleNormalGroupNode(t *testing.T) {
|
||||
+ ctx := initReconcileTestDB(t)
|
||||
+ system := db.System{Running: true}
|
||||
+ mustNoError(t, db.DB(ctx).Create(&system).Error)
|
||||
+ sub, nodes := createTestSubscription(t, ctx, "ss://current", "ss://stale")
|
||||
+ group := createTestGroup(t, ctx, "min_moving_avg", &system.ID, nodes...)
|
||||
+
|
||||
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://current"}))
|
||||
+
|
||||
+ var gotGroup db.Group
|
||||
+ mustNoError(t, db.DB(ctx).First(&gotGroup, group.ID).Error)
|
||||
+ assertEqualValues(t, 1, gotGroup.Version)
|
||||
+ assertEqualValues(t, 0, countGroupNode(t, ctx, group.ID, nodes[1].ID))
|
||||
+}
|
||||
+
|
||||
+func TestReconcileDetachesStaleFixedNode(t *testing.T) {
|
||||
+ ctx := initReconcileTestDB(t)
|
||||
+ sub, nodes := createTestSubscription(t, ctx, "ss://current", "ss://fixed-stale")
|
||||
+ group := createTestGroup(t, ctx, "fixed", nil, nodes[1])
|
||||
+
|
||||
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://current"}))
|
||||
+
|
||||
+ var got db.Node
|
||||
+ mustNoError(t, db.DB(ctx).First(&got, nodes[1].ID).Error)
|
||||
+ assertNil(t, got.SubscriptionID)
|
||||
+ assertEqualValues(t, 1, countGroupNode(t, ctx, group.ID, got.ID))
|
||||
+}
|
||||
+
|
||||
+func TestReconcilePreservesNodeReferencedByFixedAndNormalGroups(t *testing.T) {
|
||||
+ ctx := initReconcileTestDB(t)
|
||||
+ sub, nodes := createTestSubscription(t, ctx, "ss://current", "ss://shared-stale")
|
||||
+ fixed := createTestGroup(t, ctx, "fixed", nil, nodes[1])
|
||||
+ normal := createTestGroup(t, ctx, "min", nil, nodes[1])
|
||||
+
|
||||
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://current"}))
|
||||
+
|
||||
+ var got db.Node
|
||||
+ mustNoError(t, db.DB(ctx).First(&got, nodes[1].ID).Error)
|
||||
+ assertNil(t, got.SubscriptionID)
|
||||
+ assertEqualValues(t, 1, countGroupNode(t, ctx, fixed.ID, got.ID))
|
||||
+ assertEqualValues(t, 1, countGroupNode(t, ctx, normal.ID, got.ID))
|
||||
+}
|
||||
+
|
||||
+func TestReconcileAllLinksAlreadyExistSucceeds(t *testing.T) {
|
||||
+ ctx := initReconcileTestDB(t)
|
||||
+ sub, nodes := createTestSubscription(t, ctx, "ss://one", "ss://two")
|
||||
+
|
||||
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://one", "ss://two", "ss://two"}))
|
||||
+
|
||||
+ var count int64
|
||||
+ mustNoError(t, db.DB(ctx).Model(&db.Node{}).Where("subscription_id = ?", sub.ID).Count(&count).Error)
|
||||
+ assertEqualValues(t, len(nodes), count)
|
||||
+}
|
||||
+
|
||||
+func TestReconcileRollsBackWhenAnyIncomingNodeIsInvalid(t *testing.T) {
|
||||
+ ctx := initReconcileTestDB(t)
|
||||
+ sub, nodes := createTestSubscription(t, ctx, "ss://current", "ss://stale")
|
||||
+ tx := db.BeginTx(ctx)
|
||||
+
|
||||
+ assertError(t, reconcileSubscriptionNodes(tx, sub.ID, []string{"ss://current", "not-a-node-link"}))
|
||||
+ mustNoError(t, tx.Rollback().Error)
|
||||
+
|
||||
+ var count int64
|
||||
+ mustNoError(t, db.DB(ctx).Model(&db.Node{}).Where("id IN ?", []uint{nodes[0].ID, nodes[1].ID}).Count(&count).Error)
|
||||
+ assertEqualValues(t, 2, count)
|
||||
+}
|
||||
+
|
||||
+func TestUpdateSubscriptionRejectsChangedLink(t *testing.T) {
|
||||
+ ctx := initReconcileTestDB(t)
|
||||
+ sub, nodes := createTestSubscription(t, ctx, "ss://current")
|
||||
+ snapshot := sub
|
||||
+ mustNoError(t, db.DB(ctx).Model(&sub).Update("link", "https://example.invalid/changed").Error)
|
||||
+
|
||||
+ assertError(t, updateSubscriptionTx(ctx, &snapshot, sub.ID, []string{"ss://replacement"}))
|
||||
+
|
||||
+ var got db.Node
|
||||
+ mustNoError(t, db.DB(ctx).First(&got, nodes[0].ID).Error)
|
||||
+ assertNotNil(t, got.SubscriptionID)
|
||||
+ assertEqualValues(t, sub.ID, *got.SubscriptionID)
|
||||
+}
|
||||
@@ -0,0 +1,24 @@
|
||||
diff --git a/graphql/service/subscription/mutation_utils.go b/graphql/service/subscription/mutation_utils.go
|
||||
index e01a342..c6bc3e1 100644
|
||||
--- a/graphql/service/subscription/mutation_utils.go
|
||||
+++ b/graphql/service/subscription/mutation_utils.go
|
||||
@@ -171,6 +171,8 @@ func AutoUpdateVersionByIds(d *gorm.DB, ids []uint) (err error) {
|
||||
return nil
|
||||
}
|
||||
|
||||
+const scheduledUpdateTimeout = 3 * time.Minute
|
||||
+
|
||||
var (
|
||||
schedulerCache = make(map[uint]*gocron.Scheduler)
|
||||
schedulerMu sync.RWMutex
|
||||
@@ -224,7 +226,9 @@ func AddUpdateScheduler(ctx context.Context, id uint) {
|
||||
}
|
||||
logrus.Info("Subscription " + tag + " update task enabled, with exp " + sub.CronExp)
|
||||
_, err := s.Cron(sub.CronExp).Do(func() {
|
||||
- if _, err := UpdateById(context.Background(), sub.ID); err != nil {
|
||||
+ ctx, cancel := context.WithTimeout(context.Background(), scheduledUpdateTimeout)
|
||||
+ defer cancel()
|
||||
+ if _, err := UpdateById(ctx, sub.ID); err != nil {
|
||||
logrus.Error(err)
|
||||
}
|
||||
})
|
||||
@@ -0,0 +1,73 @@
|
||||
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
|
||||
Subject: [PATCH] daed: open sqlite database in WAL mode
|
||||
MIME-Version: 1.0
|
||||
Content-Type: text/plain; charset=UTF-8
|
||||
Content-Transfer-Encoding: 8bit
|
||||
|
||||
订阅自动更新写库时，WebUI 登录查询被锁 5 秒后报 database is locked。
|
||||
|
||||
开 WAL 并把 busy_timeout 提到 30 秒，读不再被写事务阻塞；WAL 附带的两个文件跟主库一样收紧到 0640。
|
||||
---
|
||||
db/db.go | 17 ++++++++++++-----
|
||||
pkg/sqlite/sqlite_mipsarch.go | 4 +++-
|
||||
pkg/sqlite/sqlite_others.go | 4 +++-
|
||||
3 files changed, 18 insertions(+), 7 deletions(-)
|
||||
|
||||
diff --git a/db/db.go b/db/db.go
|
||||
index a79022b..6145443 100644
|
||||
--- a/db/db.go
|
||||
+++ b/db/db.go
|
||||
@@ -50,13 +50,20 @@ func InitDatabase(configDir string) (err error) {
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
- if fi, err := os.Stat(path); err != nil {
|
||||
- return err
|
||||
- } else if fi.Mode()&0037 > 0 {
|
||||
- // Too open, chmod it to 0640.
|
||||
- if err = os.Chmod(path, 0640); err != nil {
|
||||
+ for _, p := range []string{path, path + "-wal", path + "-shm"} {
|
||||
+ fi, err := os.Stat(p)
|
||||
+ if os.IsNotExist(err) {
|
||||
+ continue
|
||||
+ }
|
||||
+ if err != nil {
|
||||
return err
|
||||
}
|
||||
+ if fi.Mode()&0037 > 0 {
|
||||
+ // Too open, chmod it to 0640.
|
||||
+ if err = os.Chmod(p, 0640); err != nil {
|
||||
+ return err
|
||||
+ }
|
||||
+ }
|
||||
}
|
||||
|
||||
return nil
|
||||
diff --git a/pkg/sqlite/sqlite_mipsarch.go b/pkg/sqlite/sqlite_mipsarch.go
|
||||
index 7b8c1b4..8c275e5 100644
|
||||
--- a/pkg/sqlite/sqlite_mipsarch.go
|
||||
+++ b/pkg/sqlite/sqlite_mipsarch.go
|
||||
@@ -10,6 +10,8 @@ import (
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
+const pragmas = "?_journal_mode=WAL&_busy_timeout=30000&_synchronous=FULL"
|
||||
+
|
||||
func Open(dsn string) gorm.Dialector {
|
||||
- return sqlite.Open(dsn)
|
||||
+ return sqlite.Open(dsn + pragmas)
|
||||
}
|
||||
diff --git a/pkg/sqlite/sqlite_others.go b/pkg/sqlite/sqlite_others.go
|
||||
index 4598651..83d8b80 100644
|
||||
--- a/pkg/sqlite/sqlite_others.go
|
||||
+++ b/pkg/sqlite/sqlite_others.go
|
||||
@@ -7,6 +7,8 @@ import (
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
+const pragmas = "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(30000)&_pragma=synchronous(FULL)"
|
||||
+
|
||||
func Open(dsn string) gorm.Dialector {
|
||||
- return sqlite.Open(dsn)
|
||||
+ return sqlite.Open(dsn + pragmas)
|
||||
}
|
||||
@@ -0,0 +1,166 @@
|
||||
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
|
||||
Subject: [PATCH] daed: reload dae outside the database transaction
|
||||
MIME-Version: 1.0
|
||||
Content-Type: text/plain; charset=UTF-8
|
||||
Content-Transfer-Encoding: 8bit
|
||||
|
||||
配置生效前先提交读事务，reload 完再用短事务写运行状态。
|
||||
|
||||
原来整个 reload 都在事务里，最长可占着数据库锁一百多秒；写状态时按 ID 重查仍存在的分组，避免把 reload 期间被删的分组重新插回去。
|
||||
---
|
||||
graphql/service/config/mutation_utils.go | 77 ++++++++++++++++--------
|
||||
1 file changed, 53 insertions(+), 24 deletions(-)
|
||||
|
||||
diff --git a/graphql/service/config/mutation_utils.go b/graphql/service/config/mutation_utils.go
|
||||
index feb94d3..59f0ebd 100644
|
||||
--- a/graphql/service/config/mutation_utils.go
|
||||
+++ b/graphql/service/config/mutation_utils.go
|
||||
@@ -270,28 +270,12 @@ func restartAfterReloadTimeout() {
|
||||
}()
|
||||
}
|
||||
|
||||
-func runTransaction(ctx context.Context, noLoad bool) (n int32, err error) {
|
||||
- tx := db.BeginTx(ctx)
|
||||
- if tx.Error != nil {
|
||||
- return 0, tx.Error
|
||||
- }
|
||||
- n, err = runLocked(tx, noLoad)
|
||||
- if err != nil {
|
||||
- tx.Rollback()
|
||||
- return 0, err
|
||||
- }
|
||||
- if err = tx.Commit().Error; err != nil {
|
||||
- return 0, err
|
||||
- }
|
||||
- return n, nil
|
||||
-}
|
||||
-
|
||||
func Run(ctx context.Context, noLoad bool) (n int32, err error) {
|
||||
if ok := runLock.TryLock(); !ok {
|
||||
return 0, fmt.Errorf("the last request didn't complete; make a cup of tea and take a break")
|
||||
}
|
||||
defer runLock.Unlock()
|
||||
- return runTransaction(ctx, noLoad)
|
||||
+ return runLocked(ctx, noLoad)
|
||||
}
|
||||
|
||||
func ApplyIfRunning(ctx context.Context) error {
|
||||
@@ -307,11 +291,11 @@ func ApplyIfRunning(ctx context.Context) error {
|
||||
if err != nil || !modified {
|
||||
return err
|
||||
}
|
||||
- _, err = runTransaction(ctx, false)
|
||||
+ _, err = runLocked(ctx, false)
|
||||
return err
|
||||
}
|
||||
|
||||
-func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
|
||||
+func runLocked(ctx context.Context, noLoad bool) (n int32, err error) {
|
||||
//// Dry run.
|
||||
if noLoad {
|
||||
ch := make(chan error)
|
||||
@@ -336,18 +320,38 @@ func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
|
||||
}
|
||||
|
||||
// Running -> false
|
||||
+ tx := db.BeginTx(ctx)
|
||||
+ if tx.Error != nil {
|
||||
+ return 0, tx.Error
|
||||
+ }
|
||||
var sys db.System
|
||||
- if err = d.Model(&db.System{}).FirstOrCreate(&sys).Error; err != nil {
|
||||
+ if err = tx.Model(&db.System{}).FirstOrCreate(&sys).Error; err != nil {
|
||||
+ tx.Rollback()
|
||||
return 0, err
|
||||
}
|
||||
- if err = d.Model(&sys).Updates(map[string]interface{}{
|
||||
+ if err = tx.Model(&sys).Updates(map[string]interface{}{
|
||||
"running": false,
|
||||
}).Error; err != nil {
|
||||
+ tx.Rollback()
|
||||
+ return 0, err
|
||||
+ }
|
||||
+ if err = tx.Commit().Error; err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return 1, nil
|
||||
}
|
||||
|
||||
+ d := db.BeginTx(ctx)
|
||||
+ if d.Error != nil {
|
||||
+ return 0, d.Error
|
||||
+ }
|
||||
+ readTx := d
|
||||
+ defer func() {
|
||||
+ if readTx != nil {
|
||||
+ readTx.Rollback()
|
||||
+ }
|
||||
+ }()
|
||||
+
|
||||
//// Run selected global+dns+routing.
|
||||
/// Get them from database and parse them to daeConfig.
|
||||
var mConfig db.Config
|
||||
@@ -513,6 +517,11 @@ func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
|
||||
c.Node = append(c.Node, daeConfig.KeyableString(fmt.Sprintf("%v:%v", node.uniqueName, node.dbNode.Link)))
|
||||
}
|
||||
|
||||
+ if err = readTx.Commit().Error; err != nil {
|
||||
+ return 0, err
|
||||
+ }
|
||||
+ readTx = nil
|
||||
+
|
||||
/// Reload with current config.
|
||||
chReloadCallback := make(chan error)
|
||||
reloadMsg := &dae.ReloadMessage{
|
||||
@@ -536,20 +545,37 @@ func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
|
||||
}
|
||||
|
||||
// Save running status
|
||||
+ tx := db.BeginTx(context.WithoutCancel(ctx))
|
||||
+ if tx.Error != nil {
|
||||
+ return 0, tx.Error
|
||||
+ }
|
||||
+ defer func() {
|
||||
+ if err != nil {
|
||||
+ tx.Rollback()
|
||||
+ }
|
||||
+ }()
|
||||
var sys db.System
|
||||
- if err = d.Model(&db.System{}).FirstOrCreate(&sys).Error; err != nil {
|
||||
+ if err = tx.Model(&db.System{}).FirstOrCreate(&sys).Error; err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var gvs uint
|
||||
var gids []string
|
||||
+ groupIds := make([]uint, 0, len(groups))
|
||||
for _, g := range groups {
|
||||
gvs += g.Version
|
||||
gids = append(gids, fmt.Sprintf("%x", g.ID))
|
||||
+ groupIds = append(groupIds, g.ID)
|
||||
+ }
|
||||
+ liveGroups := make([]db.Group, 0, len(groupIds))
|
||||
+ if len(groupIds) > 0 {
|
||||
+ if err = tx.Where("id IN ?", groupIds).Find(&liveGroups).Error; err != nil {
|
||||
+ return 0, err
|
||||
+ }
|
||||
}
|
||||
sort.Slice(gids, func(i, j int) bool {
|
||||
return gids[i] < gids[j]
|
||||
})
|
||||
- if err = d.Model(&sys).Updates(map[string]interface{}{
|
||||
+ if err = tx.Model(&sys).Updates(map[string]interface{}{
|
||||
"running": true,
|
||||
"running_config_id": mConfig.ID,
|
||||
"running_config_version": mConfig.Version,
|
||||
@@ -562,7 +588,10 @@ func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
|
||||
}).Error; err != nil {
|
||||
return 0, err
|
||||
}
|
||||
- if err = d.Model(&sys).Association("RunningGroups").Replace(groups); err != nil {
|
||||
+ if err = tx.Model(&sys).Association("RunningGroups").Replace(liveGroups); err != nil {
|
||||
+ return 0, err
|
||||
+ }
|
||||
+ if err = tx.Commit().Error; err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
|
||||
Subject: [PATCH] daed: serialize scheduled subscription updates
|
||||
MIME-Version: 1.0
|
||||
Content-Type: text/plain; charset=UTF-8
|
||||
Content-Transfer-Encoding: 8bit
|
||||
|
||||
订阅更新的入口读也纳入同一把锁，整轮更新串行，排到才开始计时。
|
||||
|
||||
多个订阅定时任务撞在同一分钟时会互相锁死。
|
||||
---
|
||||
.../service/subscription/mutation_utils.go | 29 +++++++++++++------
|
||||
1 file changed, 20 insertions(+), 9 deletions(-)
|
||||
|
||||
diff --git a/graphql/service/subscription/mutation_utils.go b/graphql/service/subscription/mutation_utils.go
|
||||
index c6bc3e1..e2aec32 100644
|
||||
--- a/graphql/service/subscription/mutation_utils.go
|
||||
+++ b/graphql/service/subscription/mutation_utils.go
|
||||
@@ -225,12 +225,8 @@ func AddUpdateScheduler(ctx context.Context, id uint) {
|
||||
tag = *sub.Tag
|
||||
}
|
||||
logrus.Info("Subscription " + tag + " update task enabled, with exp " + sub.CronExp)
|
||||
- _, err := s.Cron(sub.CronExp).Do(func() {
|
||||
- ctx, cancel := context.WithTimeout(context.Background(), scheduledUpdateTimeout)
|
||||
- defer cancel()
|
||||
- if _, err := UpdateById(ctx, sub.ID); err != nil {
|
||||
- logrus.Error(err)
|
||||
- }
|
||||
+ _, err := s.Cron(sub.CronExp).SingletonMode().Do(func() {
|
||||
+ runScheduledUpdate(sub.ID)
|
||||
})
|
||||
if err != nil {
|
||||
logrus.Errorf("Failed to schedule subscription %d update: invalid cron expression '%s': %v", sub.ID, sub.CronExp, err)
|
||||
@@ -364,9 +360,6 @@ func reconcileSubscriptionNodes(tx *gorm.DB, subId uint, links []string) error {
|
||||
}
|
||||
|
||||
func updateSubscriptionTx(ctx context.Context, m *db.Subscription, subId uint, links []string) (err error) {
|
||||
- subscriptionUpdateMu.Lock()
|
||||
- defer subscriptionUpdateMu.Unlock()
|
||||
-
|
||||
tx := db.BeginTx(ctx)
|
||||
if tx.Error != nil {
|
||||
return tx.Error
|
||||
@@ -404,7 +397,25 @@ func updateSubscriptionTx(ctx context.Context, m *db.Subscription, subId uint, l
|
||||
return tx.Commit().Error
|
||||
}
|
||||
|
||||
+func runScheduledUpdate(subId uint) {
|
||||
+ subscriptionUpdateMu.Lock()
|
||||
+ defer subscriptionUpdateMu.Unlock()
|
||||
+
|
||||
+ ctx, cancel := context.WithTimeout(context.Background(), scheduledUpdateTimeout)
|
||||
+ defer cancel()
|
||||
+ if _, err := updateByIdLocked(ctx, subId); err != nil {
|
||||
+ logrus.Error(err)
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
func UpdateById(ctx context.Context, subId uint) (sub *db.Subscription, err error) {
|
||||
+ subscriptionUpdateMu.Lock()
|
||||
+ defer subscriptionUpdateMu.Unlock()
|
||||
+
|
||||
+ return updateByIdLocked(ctx, subId)
|
||||
+}
|
||||
+
|
||||
+func updateByIdLocked(ctx context.Context, subId uint) (sub *db.Subscription, err error) {
|
||||
var m db.Subscription
|
||||
if err = db.DB(ctx).Where(&db.Subscription{ID: subId}).First(&m).Error; err != nil {
|
||||
return nil, err
|
||||
@@ -0,0 +1,44 @@
|
||||
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
|
||||
Subject: [PATCH] daed: apply subscription cron changes without restart
|
||||
MIME-Version: 1.0
|
||||
Content-Type: text/plain; charset=UTF-8
|
||||
Content-Transfer-Encoding: 8bit
|
||||
|
||||
改 cron 后先提交事务再重建调度器。
|
||||
|
||||
原来在事务里就去重建，调度器另开连接读到的还是旧表达式，必须重启 daed 才生效。
|
||||
---
|
||||
graphql/service/subscription/mutation_utils.go | 12 +++++++++---
|
||||
1 file changed, 9 insertions(+), 3 deletions(-)
|
||||
|
||||
diff --git a/graphql/service/subscription/mutation_utils.go b/graphql/service/subscription/mutation_utils.go
|
||||
index e2aec32..4f6bcfc 100644
|
||||
--- a/graphql/service/subscription/mutation_utils.go
|
||||
+++ b/graphql/service/subscription/mutation_utils.go
|
||||
@@ -561,10 +561,12 @@ func UpdateCron(ctx context.Context, _id graphql.ID, cronExp string, cronEnable
|
||||
}
|
||||
|
||||
tx := db.BeginTx(ctx)
|
||||
+ if tx.Error != nil {
|
||||
+ return nil, tx.Error
|
||||
+ }
|
||||
+ committed := false
|
||||
defer func() {
|
||||
- if err == nil {
|
||||
- tx.Commit()
|
||||
- } else {
|
||||
+ if !committed {
|
||||
tx.Rollback()
|
||||
}
|
||||
}()
|
||||
@@ -583,6 +585,10 @@ func UpdateCron(ctx context.Context, _id graphql.ID, cronExp string, cronEnable
|
||||
}).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
+ if err = tx.Commit().Error; err != nil {
|
||||
+ return nil, err
|
||||
+ }
|
||||
+ committed = true
|
||||
|
||||
// Update scheduler
|
||||
RemoveUpdateScheduler(id)
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,24 @@
|
||||
--- a/dae-core/trace/kern/trace.c
|
||||
+++ b/dae-core/trace/kern/trace.c
|
||||
@@ -228,7 +228,7 @@ KPROBE_SKB_AT(1)
|
||||
KPROBE_SKB_AT(2)
|
||||
KPROBE_SKB_AT(3)
|
||||
KPROBE_SKB_AT(4)
|
||||
-KPROBE_SKB_AT(5)
|
||||
+//KPROBE_SKB_AT(5)
|
||||
|
||||
SEC("kprobe/skb_lifetime_termination")
|
||||
int kprobe_skb_lifetime_termination(struct pt_regs *ctx)
|
||||
--- a/dae-core/trace/trace.go
|
||||
+++ b/dae-core/trace/trace.go
|
||||
@@ -206,8 +206,8 @@ func attachBpfToTargets(objs *bpfObjects
|
||||
kp, err = link.Kprobe(fn, objs.KprobeSkb3, nil)
|
||||
case 4:
|
||||
kp, err = link.Kprobe(fn, objs.KprobeSkb4, nil)
|
||||
- case 5:
|
||||
- kp, err = link.Kprobe(fn, objs.KprobeSkb5, nil)
|
||||
+ //case 5:
|
||||
+ // kp, err = link.Kprobe(fn, objs.KprobeSkb5, nil)
|
||||
}
|
||||
if err != nil {
|
||||
logrus.Debugf("failed to attach kprobe to %s: %+v\n", fn, err)
|
||||
Reference in new issue
Block a user