Skip to content

Commit e62a879

Browse files
PigbibiPigbibi
authored
Fix internal LongBridge precheck invoke
Route internal Cloud Run manual precheck/probe invokes through Scheduler and label precheck notifications correctly.
1 parent f6e8160 commit e62a879

10 files changed

Lines changed: 309 additions & 9 deletions

‎.github/workflows/invoke-cloud-run.yml‎

Lines changed: 191 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ jobs:
3939
env:
4040
CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }}
4141
CLOUD_RUN_SERVICE: ${{ vars.CLOUD_RUN_SERVICE }}
42+
CLOUD_SCHEDULER_LOCATION: ${{ vars.CLOUD_SCHEDULER_LOCATION }}
4243
steps:
4344
- name: Validate inputs
4445
run: |
@@ -82,20 +83,127 @@ jobs:
8283
raw_path="/${raw_path}"
8384
fi
8485
85-
service_url="$(
86+
service_json="$(
8687
gcloud run services describe "${CLOUD_RUN_SERVICE}" \
8788
--region "${CLOUD_RUN_REGION}" \
88-
--format='value(status.url)'
89+
--format=json
90+
)"
91+
service_url="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
92+
import json
93+
import os
94+
95+
service = json.loads(os.environ["SERVICE_JSON"])
96+
print((service.get("status") or {}).get("url") or "")
97+
PY
8998
)"
9099
if [ -z "${service_url}" ]; then
91100
echo "Unable to resolve Cloud Run service URL." >&2
92101
exit 1
93102
fi
94103
104+
service_ingress="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
105+
import json
106+
import os
107+
108+
service = json.loads(os.environ["SERVICE_JSON"])
109+
annotations = (service.get("metadata") or {}).get("annotations") or {}
110+
print(annotations.get("run.googleapis.com/ingress-status") or annotations.get("run.googleapis.com/ingress") or "")
111+
PY
112+
)"
113+
latest_ready_revision="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
114+
import json
115+
import os
116+
117+
service = json.loads(os.environ["SERVICE_JSON"])
118+
print((service.get("status") or {}).get("latestReadyRevisionName") or "")
119+
PY
120+
)"
121+
deployed_commit="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
122+
import json
123+
import os
124+
125+
service = json.loads(os.environ["SERVICE_JSON"])
126+
template = ((service.get("spec") or {}).get("template") or {}).get("metadata") or {}
127+
print((template.get("labels") or {}).get("commit-sha") or "")
128+
PY
129+
)"
130+
131+
invoke_method="direct"
132+
scheduler_job=""
133+
scheduler_location=""
134+
if [ "${service_ingress}" = "internal" ]; then
135+
scheduler_location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}"
136+
case "${raw_path}" in
137+
/)
138+
scheduler_job="${CLOUD_RUN_SERVICE}-scheduler"
139+
;;
140+
/probe)
141+
scheduler_job="${CLOUD_RUN_SERVICE}-probe-scheduler"
142+
;;
143+
/precheck)
144+
scheduler_job="${CLOUD_RUN_SERVICE}-precheck-scheduler"
145+
;;
146+
*)
147+
echo "Cloud Run service ${CLOUD_RUN_SERVICE} has internal ingress, so GitHub-hosted runners cannot curl ${raw_path} directly." >&2
148+
echo "Use one of the scheduler-backed paths: /, /probe, /precheck." >&2
149+
exit 1
150+
;;
151+
esac
152+
153+
scheduler_uri="$(
154+
gcloud scheduler jobs describe "${scheduler_job}" \
155+
--location="${scheduler_location}" \
156+
--format='value(httpTarget.uri)' 2>/dev/null || true
157+
)"
158+
if [ -z "${scheduler_uri}" ]; then
159+
echo "Cloud Scheduler job ${scheduler_job} was not found in ${scheduler_location}." >&2
160+
exit 1
161+
fi
162+
scheduler_path="$(SCHEDULER_URI="${scheduler_uri}" python3 - <<'PY'
163+
import os
164+
from urllib.parse import urlparse
165+
166+
def normalize(path: str) -> str:
167+
clean = (path or "/").rstrip("/")
168+
return clean or "/"
169+
170+
print(normalize(urlparse(os.environ["SCHEDULER_URI"]).path))
171+
PY
172+
)"
173+
requested_path="$(RAW_PATH="${raw_path}" python3 - <<'PY'
174+
import os
175+
176+
clean = (os.environ["RAW_PATH"] or "/").rstrip("/")
177+
print(clean or "/")
178+
PY
179+
)"
180+
if [ "${scheduler_path}" != "${requested_path}" ]; then
181+
echo "Cloud Scheduler job ${scheduler_job} targets ${scheduler_uri}, not ${raw_path}." >&2
182+
exit 1
183+
fi
184+
invoke_method="scheduler"
185+
fi
186+
187+
echo "Cloud Run service: ${CLOUD_RUN_SERVICE}"
188+
echo "Cloud Run region: ${CLOUD_RUN_REGION}"
189+
echo "Cloud Run URL: ${service_url}"
190+
echo "Cloud Run ingress: ${service_ingress:-<empty>}"
191+
echo "Latest ready revision: ${latest_ready_revision:-<empty>}"
192+
echo "Deployed commit: ${deployed_commit:-<empty>}"
193+
echo "Invoke method: ${invoke_method}"
194+
if [ -n "${scheduler_job}" ]; then
195+
echo "Cloud Scheduler job: ${scheduler_job}"
196+
echo "Cloud Scheduler location: ${scheduler_location}"
197+
fi
198+
95199
echo "url=${service_url}" >> "$GITHUB_OUTPUT"
96200
echo "path=${raw_path}" >> "$GITHUB_OUTPUT"
201+
echo "invoke_method=${invoke_method}" >> "$GITHUB_OUTPUT"
202+
echo "scheduler_job=${scheduler_job}" >> "$GITHUB_OUTPUT"
203+
echo "scheduler_location=${scheduler_location}" >> "$GITHUB_OUTPUT"
97204
98205
- name: Authenticate for service invocation
206+
if: steps.service.outputs.invoke_method == 'direct'
99207
id: invoke-auth
100208
uses: google-github-actions/auth@v3
101209
with:
@@ -106,10 +214,91 @@ jobs:
106214
id_token_include_email: true
107215

108216
- name: Invoke service
217+
if: steps.service.outputs.invoke_method == 'direct'
109218
run: |
110219
set -euo pipefail
111220
112221
curl --fail-with-body --show-error --silent \
113222
--request POST \
114223
--header "Authorization: Bearer ${{ steps.invoke-auth.outputs.id_token }}" \
115224
"${{ steps.service.outputs.url }}${{ steps.service.outputs.path }}"
225+
226+
- name: Invoke internal service through Cloud Scheduler
227+
if: steps.service.outputs.invoke_method == 'scheduler'
228+
run: |
229+
set -euo pipefail
230+
231+
started_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
232+
scheduler_job="${{ steps.service.outputs.scheduler_job }}"
233+
scheduler_location="${{ steps.service.outputs.scheduler_location }}"
234+
235+
echo "Triggering ${scheduler_job} at ${started_at}."
236+
gcloud scheduler jobs run "${scheduler_job}" \
237+
--location="${scheduler_location}" \
238+
--quiet
239+
240+
deadline=$((SECONDS + 180))
241+
while true; do
242+
job_json="$(
243+
gcloud scheduler jobs describe "${scheduler_job}" \
244+
--location="${scheduler_location}" \
245+
--format=json
246+
)"
247+
attempt_seen="$(JOB_JSON="${job_json}" STARTED_AT="${started_at}" python3 - <<'PY'
248+
import datetime as dt
249+
import json
250+
import os
251+
252+
def parse_timestamp(value: str) -> dt.datetime | None:
253+
if not value:
254+
return None
255+
text = value.replace("Z", "+00:00")
256+
return dt.datetime.fromisoformat(text)
257+
258+
job = json.loads(os.environ["JOB_JSON"])
259+
last_attempt = parse_timestamp(job.get("lastAttemptTime") or "")
260+
started_at = parse_timestamp(os.environ["STARTED_AT"])
261+
print("true" if last_attempt and started_at and last_attempt >= started_at else "false")
262+
PY
263+
)"
264+
status_code="$(JOB_JSON="${job_json}" python3 - <<'PY'
265+
import json
266+
import os
267+
268+
status = (json.loads(os.environ["JOB_JSON"]).get("status") or {})
269+
print(status.get("code") or "")
270+
PY
271+
)"
272+
status_message="$(JOB_JSON="${job_json}" python3 - <<'PY'
273+
import json
274+
import os
275+
276+
status = (json.loads(os.environ["JOB_JSON"]).get("status") or {})
277+
print(status.get("message") or "")
278+
PY
279+
)"
280+
last_attempt_time="$(JOB_JSON="${job_json}" python3 - <<'PY'
281+
import json
282+
import os
283+
284+
print(json.loads(os.environ["JOB_JSON"]).get("lastAttemptTime") or "")
285+
PY
286+
)"
287+
288+
if [ "${attempt_seen}" = "true" ]; then
289+
if [ -n "${status_code}" ] && [ "${status_code}" != "0" ]; then
290+
echo "Cloud Scheduler job ${scheduler_job} failed with status ${status_code}: ${status_message}" >&2
291+
exit 1
292+
fi
293+
echo "Cloud Scheduler job ${scheduler_job} ran at ${last_attempt_time}."
294+
break
295+
fi
296+
297+
if [ "${SECONDS}" -ge "${deadline}" ]; then
298+
echo "Timed out waiting for Cloud Scheduler job ${scheduler_job} to record a new attempt." >&2
299+
exit 1
300+
fi
301+
302+
echo "Waiting for Cloud Scheduler job ${scheduler_job} attempt..."
303+
sleep 10
304+
done

‎application/rebalance_service.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -371,6 +371,7 @@ def fetch_replanned_state():
371371
strategy_display_name=config.strategy_display_name,
372372
dry_run_only=config.dry_run_only,
373373
extra_notification_lines=config.extra_notification_lines,
374+
title_key=config.notification_title_key or "rebalance_title",
374375
)
375376
)
376377
else:
@@ -384,6 +385,7 @@ def fetch_replanned_state():
384385
strategy_display_name=config.strategy_display_name,
385386
dry_run_only=config.dry_run_only,
386387
extra_notification_lines=config.extra_notification_lines,
388+
title_key=config.notification_title_key or "heartbeat_title",
387389
)
388390
)
389391
return execution_result

‎application/runtime_composer.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -196,6 +196,7 @@ def build_rebalance_config(
196196
*,
197197
strategy_plugin_signals=(),
198198
strategy_plugin_error: str | None = None,
199+
notification_title_key: str = "",
199200
) -> LongBridgeRebalanceConfig:
200201
market_scope_line = self.translator(
201202
"market_scope_detail",
@@ -231,6 +232,7 @@ def build_rebalance_config(
231232
safe_haven_cash_substitute_threshold_usd=self.safe_haven_cash_substitute_threshold_usd,
232233
sleeper=self.sleeper,
233234
extra_notification_lines=(market_scope_line, *plugin_lines, *plugin_error_lines),
235+
notification_title_key=notification_title_key,
234236
strategy_plugin_signals=tuple(strategy_plugin_signals or ()),
235237
execution_dedup_enabled=resolve_execution_dedup_enabled(
236238
env_reader=self.env_reader,

‎application/runtime_dependencies.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ class LongBridgeRebalanceConfig:
2626
safe_haven_cash_substitute_threshold_usd: float = 1000.0
2727
sleeper: Callable[[float], None] | None = None
2828
extra_notification_lines: tuple[str, ...] = ()
29+
notification_title_key: str = ""
2930
strategy_plugin_signals: tuple[Any, ...] = ()
3031
execution_dedup_enabled: bool = False
3132
execution_state_store: Any = None

‎main.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -505,6 +505,11 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
505505
config=composer.build_rebalance_config(
506506
strategy_plugin_signals=strategy_plugin_signals,
507507
strategy_plugin_error=strategy_plugin_error,
508+
notification_title_key=(
509+
"precheck_title"
510+
if validation_only and validation_label == "precheck"
511+
else ""
512+
),
508513
),
509514
)
510515
signal_snapshot = {}

‎notifications/renderers.py‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -506,9 +506,10 @@ def render_rebalance_notification(
506506
strategy_display_name,
507507
dry_run_only,
508508
extra_notification_lines=(),
509+
title_key="rebalance_title",
509510
) -> RenderedNotification:
510511
formatted_logs = "\n".join(f" - {log}" for log in [*logs, *skip_logs, *note_logs])
511-
detailed_lines = [translator("rebalance_title")]
512+
detailed_lines = [translator(title_key or "rebalance_title")]
512513
_append_strategy_line(detailed_lines, strategy_display_name=strategy_display_name, translator=translator)
513514
if dry_run_only:
514515
detailed_lines.append(translator("dry_run_banner"))
@@ -525,7 +526,7 @@ def render_rebalance_notification(
525526
)
526527
detailed_lines.extend([separator, translator("order_logs_title"), formatted_logs])
527528

528-
compact_lines = [translator("rebalance_title")]
529+
compact_lines = [translator(title_key or "rebalance_title")]
529530
_append_strategy_line(compact_lines, strategy_display_name=strategy_display_name, translator=translator)
530531
if dry_run_only:
531532
compact_lines.append(translator("dry_run_banner"))
@@ -557,8 +558,9 @@ def render_heartbeat_notification(
557558
strategy_display_name,
558559
dry_run_only,
559560
extra_notification_lines=(),
561+
title_key="heartbeat_title",
560562
) -> RenderedNotification:
561-
detailed_lines = [translator("heartbeat_title")]
563+
detailed_lines = [translator(title_key or "heartbeat_title")]
562564
_append_strategy_line(detailed_lines, strategy_display_name=strategy_display_name, translator=translator)
563565
if dry_run_only:
564566
detailed_lines.append(translator("dry_run_banner"))
@@ -594,7 +596,7 @@ def render_heartbeat_notification(
594596
+ "\n".join(f" - {log}" for log in note_logs)
595597
)
596598

597-
compact_lines = [translator("heartbeat_title")]
599+
compact_lines = [translator(title_key or "heartbeat_title")]
598600
_append_strategy_line(compact_lines, strategy_display_name=strategy_display_name, translator=translator)
599601
if dry_run_only:
600602
compact_lines.append(translator("dry_run_banner"))

‎notifications/telegram.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
"income_locked": "🏦 收入层锁定占比: {ratio}",
4040
"signal": "🎯 触发信号: {msg}",
4141
"heartbeat_title": "💓 【心跳检测】",
42+
"precheck_title": "🧪 【策略预检】",
4243
"health_probe_title": "🔎 【连接探针】",
4344
"health_probe_error_prefix": "健康探针异常:\n",
4445
"equity": "💰 净值: ${value}",
@@ -196,6 +197,7 @@
196197
"income_locked": "🏦 Income Locked: {ratio}",
197198
"signal": "🎯 Signal: {msg}",
198199
"heartbeat_title": "💓 【Heartbeat】",
200+
"precheck_title": "🧪 【Strategy Precheck】",
199201
"health_probe_title": "🔎 【Health Probe】",
200202
"health_probe_error_prefix": "Health probe error:\n",
201203
"equity": "💰 Equity: ${value}",

‎tests/test_invoke_cloud_run_workflow.sh‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,11 +12,18 @@ grep -Fq "google-github-actions/auth@v3" "$workflow_file"
1212
grep -Fq "google-github-actions/setup-gcloud@v3" "$workflow_file"
1313
grep -Fq "CLOUD_RUN_REGION: \${{ vars.CLOUD_RUN_REGION }}" "$workflow_file"
1414
grep -Fq "CLOUD_RUN_SERVICE: \${{ vars.CLOUD_RUN_SERVICE }}" "$workflow_file"
15+
grep -Fq "CLOUD_SCHEDULER_LOCATION: \${{ vars.CLOUD_SCHEDULER_LOCATION }}" "$workflow_file"
1516
grep -Fq "longbridge-hk|longbridge-paper|longbridge-sg" "$workflow_file"
1617
grep -Fq "gcloud run services describe \"\${CLOUD_RUN_SERVICE}\"" "$workflow_file"
18+
grep -Fq "Cloud Run service \${CLOUD_RUN_SERVICE} has internal ingress" "$workflow_file"
19+
grep -Fq "Use one of the scheduler-backed paths: /, /probe, /precheck." "$workflow_file"
20+
grep -Fq "scheduler_job=\"\${CLOUD_RUN_SERVICE}-precheck-scheduler\"" "$workflow_file"
21+
grep -Fq "Invoke internal service through Cloud Scheduler" "$workflow_file"
22+
grep -Fq "gcloud scheduler jobs run \"\${scheduler_job}\"" "$workflow_file"
1723
grep -Fq "token_format: id_token" "$workflow_file"
1824
grep -Fq "id_token_audience: \${{ steps.service.outputs.url }}" "$workflow_file"
1925
grep -Fq "id_token_include_email: true" "$workflow_file"
26+
grep -Fq "if: steps.service.outputs.invoke_method == 'direct'" "$workflow_file"
2027
grep -Fq "curl --fail-with-body --show-error --silent" "$workflow_file"
2128
grep -Fq -- "--request POST" "$workflow_file"
2229
grep -Fq "steps.invoke-auth.outputs.id_token" "$workflow_file"

‎tests/test_notifications.py‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -177,6 +177,37 @@ def test_heartbeat_signal_snapshot_localizes_price_source(self):
177177
self.assertIn("📊 市场状态: 🚀 风险开启（SOXX+SOXL）", rendered.compact_text)
178178
self.assertNotIn("longbridge_candlesticks", rendered.compact_text)
179179

180+
def test_precheck_heartbeat_uses_precheck_title(self):
181+
rendered = render_heartbeat_notification(
182+
execution={
183+
"signal_display": "🚀 入场信号 | 原因：QQQ 高于 MA200",
184+
},
185+
skip_logs=(),
186+
note_logs=(),
187+
translator=build_translator("zh"),
188+
separator="━━━━━━━━━━━━━━━━━━",
189+
strategy_display_name="TQQQ 增长收益",
190+
dry_run_only=True,
191+
title_key="precheck_title",
192+
)
193+
en_rendered = render_heartbeat_notification(
194+
execution={
195+
"signal_display": "Entry signal | reason: QQQ is above MA200",
196+
},
197+
skip_logs=(),
198+
note_logs=(),
199+
translator=build_translator("en"),
200+
separator="━━━━━━━━━━━━━━━━━━",
201+
strategy_display_name="TQQQ Growth Income",
202+
dry_run_only=True,
203+
title_key="precheck_title",
204+
)
205+
206+
self.assertIn("🧪 【策略预检】", rendered.compact_text)
207+
self.assertNotIn("💓 【心跳检测】", rendered.compact_text)
208+
self.assertIn("🧪 【Strategy Precheck】", en_rendered.compact_text)
209+
self.assertNotIn("💓 【Heartbeat】", en_rendered.compact_text)
210+
180211
def test_heartbeat_renders_tqqq_volatility_delever_risk_control(self):
181212
zh_rendered = render_heartbeat_notification(
182213
execution={

0 commit comments

Comments
 (0)