
Huawei Cloud Functiongraph Trigger Create
- 90 installs
- 19 repo stars
- Updated July 31, 2026
- huaweicloud/huaweicloud-skills
Create scheduled TIMER triggers for Huawei Cloud FunctionGraph functions using Quartz Cron expressions for periodic execution.
About
Creates and configures scheduled TIMER triggers for FunctionGraph functions using Quartz Cron expressions. A developer uses it to run serverless functions on a schedule for periodic data processing, backups, or health checks.
- TIMER triggers with Quartz Cron scheduling
- Ideal for periodic tasks, backups, and monitoring
Huawei Cloud Functiongraph Trigger Create by the numbers
- 90 all-time installs (skills.sh)
- +17 installs in the week ending Aug 2, 2026 (Skillselion tracking)
- Ranked #588 of 1,042 Cloud & Infrastructure skills by installs in the Skillselion catalog
- Data as of Aug 2, 2026 (Skillselion catalog sync)
npx skills add https://github.com/huaweicloud/huaweicloud-skills --skill huawei-cloud-functiongraph-trigger-createAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 90 |
|---|---|
| repo stars | ★ 19 |
| Last updated | July 31, 2026 |
| Repository | huaweicloud/huaweicloud-skills ↗ |
What it does
Create scheduled TIMER triggers for Huawei Cloud FunctionGraph functions using Quartz Cron expressions for periodic execution.
Files
Overview
This skill enables the creation and configuration of scheduled triggers (TIMER type) for Huawei Cloud FunctionGraph functions. It supports Quartz Cron expression format for flexible scheduling configurations.
The trigger allows automatic execution of serverless functions at specified time intervals, making it ideal for:
- Periodic data processing tasks
- Scheduled backup operations
- Regular monitoring and health checks
- Time-based notification systems
Prerequisites
Before using this skill, ensure the following requirements are met:
1. Python Environment: Python 3.9+ installed 2. SDK Installation: Install FunctionGraph SDK: pip install huaweicloudsdkfunctiongraph 3. Environment Variables: Configure the following environment variables:
HUAWEI_AK: Huawei Cloud Access KeyHUAWEI_SK: Huawei Cloud Secret KeyHUAWEI_REGION: Target region (e.g.,cn-north-4)HUAWEI_PROJECT_ID: Project ID
4. FunctionGraph Function: Target function must already exist in the specified region 5. Network Access: Stable network connection to Huawei Cloud API endpoints
Usage
Basic Command Structure
cd scripts
python create_trigger.py \
--function-urn "urn:fss:cn-north-4:project_id:function:default:my-function:latest" \
--name "daily-trigger" \
--schedule "0 0 2 * * ?" \
--schedule-type "Cron" \
--status "ACTIVE"Examples
Cron Expression Trigger
cd scripts
python create_trigger.py \
--function-urn "urn:fss:cn-north-4:project_id:function:default:my-function:latest" \
--name "daily-trigger" \
--schedule "0 0 8 * * ?" \
--schedule-type "Cron"Fixed Rate Trigger
cd scripts
python create_trigger.py \
--function-urn "urn:fss:cn-north-4:project_id:function:default:my-function:latest" \
--name "every-5min" \
--schedule "5m" \
--schedule-type "Rate"Cron Expression Format
FunctionGraph uses Quartz Cron format with 6 or 7 fields:
┌───────────── second (0-59)
│ ┌───────────── minute (0-59)
│ │ ┌───────────── hour (0-23)
│ │ │ ┌───────────── day of month (1-31)
│ │ │ │ ┌───────────── month (1-12)
│ │ │ │ │ ┌───────────── day of week (1-7, 1=Sunday)
│ │ │ │ │ │
* * * * * ?For detailed Cron expression reference including special characters and common examples, see Cron Expression Reference.
Parameters Confirmation
Before creating the trigger, confirm the following parameters:
| Parameter | Required | Description | Example |
|---|---|---|---|
function_urn | Yes | Target function URN | urn:fss:cn-north-4:xxx:function:default:my-func:latest |
trigger_name | Yes | Trigger name (1-64 chars) | daily-trigger |
schedule | Yes | Cron expression or Rate value | 0 0 2 * * ? or 5m |
schedule_type | No | Cron (default) or Rate | Cron |
enable_status | No | ACTIVE (default) or DISABLED | ACTIVE |
user_event | No | Additional user event data | optional info |
Confirmation Checklist
- [ ] Function URN is correct and function exists
- [ ] Cron expression has been validated
- [ ] Trigger name follows naming conventions
- [ ] IAM permissions are sufficient
- [ ] Region matches the function location
Output Format
Success Response
{
"status": "success",
"trigger_id": "timer-xxx-xxx-xxx",
"trigger_name": "daily-trigger",
"trigger_type": "TIMER",
"schedule": "0 0 2 * * ?",
"enable_status": "ACTIVE",
"message": "Trigger created successfully"
}Error Response
{
"status": "failed",
"error_code": "TriggerAlreadyExists",
"message": "Trigger with the same name already exists"
}For complete error codes and troubleshooting information, see Cron Expression Reference.
Verification Method
After creating the trigger, verify the configuration:
1. List Function Triggers
Navigate to FunctionGraph console → Function details → Triggers tab to view all triggers
2. Check Trigger Details
Navigate to FunctionGraph console → Function details → Triggers → View trigger details
3. Monitor Trigger Execution
Navigate to FunctionGraph console → Function details → Triggers → View execution history
Expected indicators:
- Trigger status: Active
- Next execution time: Correctly calculated
- Execution history: No failed invocations
Best Practices
Cron Expression Best Practices
1. Avoid frequent executions: Use intervals ≥ 5 minutes unless necessary 2. Consider timezone: FunctionGraph uses UTC by default 3. Use `?` for unused day field: Either day-of-month or day-of-week should use ? 4. Validate before creation: Test expressions using online Cron validators
Naming Conventions
- Use descriptive names:
daily-data-sync,hourly-health-check - Follow pattern:
[frequency]-[purpose] - Maximum 64 characters
- Use lowercase with hyphens
Security Considerations
1. Least privilege: Grant minimal IAM permissions 2. Enable encryption: Use KMS for sensitive function inputs 3. Monitor executions: Set up Cloud Eye alarms for failures 4. Rate limiting: Configure appropriate retry parameters
Operational Recommendations
1. Start with disabled status for testing 2. Document trigger purpose in description field 3. Set appropriate retries for transient failures 4. Monitor first executions after enabling
Reference Documents
For detailed information, refer to:
- SDK Installation Guide
- IAM Policies
- Verification Method
- Acceptance Criteria
- Cron Expression Reference
Important Notes
Compatibility Notes
This skill is designed to work with Huawei Cloud FunctionGraph API. The tags field helps with skill discovery and categorization, while the version field follows semantic versioning for skill updates.
Limitations
1. Maximum triggers: Each function supports up to 10 triggers by default 2. Cron precision: Second-level scheduling may have minor delays 3. Timeout handling: Function timeout should be less than trigger interval 4. Cold start: First executions may have additional latency
Cost Implications
- No additional cost for trigger configuration
- Function execution billed per invocation
- Consider execution frequency for cost optimization
Common Pitfalls
1. Wrong timezone: Remember FunctionGraph uses UTC 2. Overlapping schedules: Multiple triggers may cause concurrent executions 3. Long-running functions: Ensure function completes before next trigger 4. Cron syntax errors: Validate expression format before deployment
For troubleshooting and error codes, refer to Cron Expression Reference.
---
Related Skills:
huawei-cloud-functiongraph-function-createhuawei-cloud-functiongraph-deploy
Acceptance Criteria
Overview
This document defines the acceptance criteria for FunctionGraph scheduled trigger configuration. All criteria must be met for successful deployment.
Functional Requirements
FR-01: Trigger Creation
Criterion: Trigger must be successfully created with correct type
Validation:
- [ ] API returns success status (HTTP 201)
- [ ] Response contains valid
trigger_id - [ ]
trigger_typeequalsTIMER - [ ] Trigger appears in function trigger list
Test Command:
hcloud functiongraph v2 list-function-triggers \
--function-urn "urn:fss:..." | grep -A 5 "TIMER"FR-02: Cron Expression Configuration
Criterion: Cron expression must be correctly stored and parsed
Validation:
- [ ]
trigger_config.schedulematches input expression - [ ] Expression is valid Quartz Cron format
- [ ] All 6-7 fields are present
- [ ] Next execution time is calculable
Test Cases:
| Input Expression | Expected Behavior | Valid |
|---|---|---|
0 0 2 * * ? | Daily at 2:00 AM | ✓ |
0 */30 * * * ? | Every 30 minutes | ✓ |
0 0 9 ? * MON-FRI | Weekdays at 9:00 AM | ✓ |
0 0 * * * | Invalid (missing field) | ✗ |
60 0 0 * * ? | Invalid (second > 59) | ✗ |
FR-03: Trigger Activation
Criterion: Trigger activation status must match configuration
Validation:
- [ ]
enable_statusequals input value - [ ] Active triggers execute per schedule
- [ ] Disabled triggers do not execute
Test Steps:
1. Create trigger with status=active 2. Verify enable_status is active 3. Wait for one scheduled execution 4. Confirm execution in history
FR-04: Function Association
Criterion: Trigger must be correctly associated with target function
Validation:
- [ ] Trigger is linked to correct function URN
- [ ] Function exists and is accessible
- [ ] Trigger appears in function's trigger list
FR-05: Naming Conventions
Criterion: Trigger name must follow naming standards
Validation:
- [ ] Name length: 1-64 characters
- [ ] Name is unique within function
- [ ] Name follows pattern:
[frequency]-[purpose]
Valid Examples:
daily-backuphourly-health-checkweekly-report-generation
Invalid Examples:
- `` (empty)
a* 65 (65 characters)Daily Backup(contains spaces, though allowed)
Non-Functional Requirements
NFR-01: Response Time
Criterion: Trigger creation must complete within acceptable time
Threshold: < 5 seconds for API response
Measurement:
time hcloud functiongraph v2 create-function-trigger ...NFR-02: Error Handling
Criterion: All errors must return meaningful messages
Validation:
- [ ] Invalid Cron returns
InvalidParameterwith details - [ ] Missing function returns
FunctionNotFound - [ ] Duplicate name returns
TriggerAlreadyExists - [ ] Permission denied returns
AccessDenied
NFR-03: Idempotency
Criterion: Re-running with same parameters should not cause errors
Validation:
- [ ] Duplicate creation returns
TriggerAlreadyExists - [ ] Existing trigger is not modified
- [ ] No duplicate triggers created
NFR-04: Security
Criterion: Operations must enforce IAM permissions
Validation:
- [ ] Operations fail without required permissions
- [ ] Audit logs capture trigger operations
- [ ] No credential exposure in logs
Integration Requirements
IR-01: Cloud Eye Integration
Criterion: Function must have monitoring capability
Validation:
- [ ] Executions appear in Cloud Eye metrics
- [ ] Alarm rules can be configured
- [ ] Error notifications are sent
IR-02: LTS Integration
Criterion: Function logs must be accessible
Validation:
- [ ] Execution logs appear in LTS
- [ ] Log stream exists for function
- [ ] Logs are queryable
IR-03: Event Flow
Criterion: Trigger-to-function event flow must work
Validation:
- [ ] Trigger event reaches function
- [ ] Event payload contains trigger metadata
- [ ] Function receives correct event structure
Operational Requirements
OR-01: Documentation
Criterion: Trigger must have adequate documentation
Validation:
- [ ] Description field is populated
- [ ] Purpose is clearly stated
- [ ] Cron expression is documented
OR-02: Monitoring
Criterion: Trigger execution must be monitorable
Validation:
- [ ] Execution history is available
- [ ] Success/failure metrics exist
- [ ] Alerting is configured
OR-03: Backup/Recovery
Criterion: Trigger configuration must be recoverable
Validation:
- [ ] Trigger details are retrievable via API
- [ ] Configuration can be exported
- [ ] Recreation from backup works
Acceptance Test Procedure
Pre-Test Setup
# Set environment variables
export HUAWEI_AK="your-ak"
export HUAWEI_SK="your-sk"
export HUAWEI_REGION="cn-north-4"
export HUAWEI_PROJECT_ID="your-project-id"
# Verify CLI is working
hcloud functiongraph v2 list-functions --limit 1Test Suite
#!/usr/bin/env python3
"""Acceptance test suite for FunctionGraph trigger creation"""
import unittest
import json
class TestTriggerAcceptance(unittest.TestCase):
def setUp(self):
# Initialize test client
pass
def test_fr01_trigger_creation(self):
"""FR-01: Trigger must be created successfully"""
# Create trigger
# Verify trigger_id is returned
# Verify trigger_type is TIMER
pass
def test_fr02_cron_configuration(self):
"""FR-02: Cron expression must be valid"""
# Test valid expressions
# Test invalid expressions
pass
def test_fr03_trigger_activation(self):
"""FR-03: Trigger status must match configuration"""
# Create with active status
# Verify status
pass
def test_fr04_function_association(self):
"""FR-04: Trigger must be linked to function"""
# Verify function URN
# Check trigger list
pass
def test_fr05_naming_conventions(self):
"""FR-05: Name must follow conventions"""
# Test valid names
# Test invalid names
pass
def test_nfr01_response_time(self):
"""NFR-01: Response must be timely"""
# Measure response time
# Assert < 5 seconds
pass
def test_nfr02_error_handling(self):
"""NFR-02: Errors must be meaningful"""
# Test various error conditions
# Verify error messages
pass
if __name__ == '__main__':
unittest.main()Post-Test Cleanup
# Remove test triggers
hcloud functiongraph v2 delete-function-trigger \
--function-urn "..." \
--trigger-id "test-trigger-id"
# Verify removal
hcloud functiongraph v2 list-function-triggers \
--function-urn "..."Acceptance Sign-off
Approval Checklist
| Criterion | Status | Approver | Date |
|---|---|---|---|
| FR-01: Trigger Creation | ☐ | ||
| FR-02: Cron Configuration | ☐ | ||
| FR-03: Trigger Activation | ☐ | ||
| FR-04: Function Association | ☐ | ||
| FR-05: Naming Conventions | ☐ | ||
| NFR-01: Response Time | ☐ | ||
| NFR-02: Error Handling | ☐ | ||
| NFR-03: Idempotency | ☐ | ||
| NFR-04: Security | ☐ | ||
| IR-01: Cloud Eye | ☐ | ||
| IR-02: LTS | ☐ | ||
| IR-03: Event Flow | ☐ | ||
| OR-01: Documentation | ☐ | ||
| OR-02: Monitoring | ☐ | ||
| OR-03: Backup/Recovery | ☐ |
Final Approval
- Test Date: _______________
- Tester: _______________
- Approver: _______________
- Status: ☐ PASS / ☐ FAIL
- Sign-off Date: _______________
Regression Test Triggers
Perform regression testing when:
1. FunctionGraph API version changes 2. Huawei Cloud SDK updates 3. IAM policy modifications 4. Network configuration changes 5. Function code updates
Related Documentation
- Verification Method
- IAM Policies
- CLI Installation Guide
Cron Expression Reference
Special Characters
| Character | Description | Example |
|---|---|---|
* | Any value | * * * * * ? = every second |
? | No specific value (for day fields) | 0 0 0 * * ? = daily at midnight |
- | Range | 0 0 9-17 * * ? = hourly 9AM-5PM |
, | List | 0 0 9,12,15 * * ? = 9AM, 12PM, 3PM |
/ | Step | 0 */5 * * * ? = every 5 minutes |
Common Examples
| Expression | Description |
|---|---|
0 0 2 * * ? | Daily at 2:00 AM |
0 */30 * * * ? | Every 30 minutes |
0 0 0 1 * ? | First day of each month at midnight |
0 0 9 ? * MON-FRI | Weekdays at 9:00 AM |
0 0 12 1,15 * ? | 1st and 15th of each month at noon |
Parameter Reference
| Parameter | Required | Description | Example |
|---|---|---|---|
function_urn | Yes | Target function URN | urn:fss:cn-north-4:xxx:function:default:my-func:latest |
trigger_name | Yes | Trigger name (1-64 chars) | daily-trigger |
schedule | Yes | Cron expression or Rate value | 0 0 2 * * ? or 5m |
schedule_type | No | Cron (default) or Rate | Cron |
enable_status | No | ACTIVE (default) or DISABLED | ACTIVE |
user_event | No | Additional user event data | optional info |
Common Error Codes
| Error Code | Description | Solution |
|---|---|---|
InvalidParameter | Invalid parameter format | Validate Cron expression and parameter values |
FunctionNotFound | Target function not found | Verify function URN and region |
TriggerAlreadyExists | Trigger name conflict | Use a different trigger name |
TriggerLimitExceeded | Maximum triggers reached | Delete unused triggers |
AccessDenied | IAM permission denied | Add required IAM policies |
Troubleshooting
| Issue | Possible Cause | Solution |
|---|---|---|
| Trigger not firing | Status is disabled | Enable the trigger |
| Execution failures | Function runtime error | Check function logs |
| Permission denied | IAM policy missing | Add required permissions |
| Invalid Cron | Syntax error | Validate expression format |
Cron Expression Format
FunctionGraph uses Quartz Cron format with 6 or 7 fields:
┌───────────── second (0-59)
│ ┌───────────── minute (0-59)
│ │ ┌───────────── hour (0-23)
│ │ │ ┌───────────── day of month (1-31)
│ │ │ │ ┌───────────── month (1-12)
│ │ │ │ │ ┌───────────── day of week (1-7, 1=Sunday)
│ │ │ │ │ │
* * * * * ?IAM Policies for FunctionGraph Trigger Management
Overview
This document specifies the minimum IAM (Identity and Access Management) permissions required to create and manage FunctionGraph scheduled triggers.
Required Permissions
Minimum Policy for Trigger Creation
{
"Version": "1.1",
"Statement": [
{
"Effect": "Allow",
"Action": [
"functiongraph:function:get",
"functiongraph:trigger:create",
"functiongraph:trigger:list",
"functiongraph:trigger:get"
],
"Resource": [
"urn:fss:*:*:function:*"
]
}
]
}Full Trigger Management Policy
For complete trigger lifecycle management, use this policy:
{
"Version": "1.1",
"Statement": [
{
"Effect": "Allow",
"Action": [
"functiongraph:function:get",
"functiongraph:function:list",
"functiongraph:trigger:create",
"functiongraph:trigger:get",
"functiongraph:trigger:list",
"functiongraph:trigger:update",
"functiongraph:trigger:delete"
],
"Resource": [
"urn:fss:*:*:function:*"
]
}
]
}Permission Actions Reference
| Action | Description | Required For |
|---|---|---|
functiongraph:function:get | Query function details | Pre-check function existence |
functiongraph:function:list | List all functions | Function discovery |
functiongraph:trigger:create | Create trigger | Core operation |
functiongraph:trigger:get | Query trigger details | Verification |
functiongraph:trigger:list | List function triggers | Verification |
functiongraph:trigger:update | Update trigger | Enable/disable trigger |
functiongraph:trigger:delete | Delete trigger | Cleanup |
Policy Assignment Methods
Method 1: Through Console
1. Navigate to IAM Console → Policies 2. Click Create Custom Policy 3. Select JSON view 4. Paste the policy JSON 5. Click OK to create 6. Attach policy to user/group/role
Method 2: Using KooCLI
# Create policy
hcloud iam v3 create-custom-policy \
--body '{
"policy": {
"name": "FunctionGraphTriggerManager",
"description": "Policy for managing FunctionGraph triggers",
"policy_document": {
"Version": "1.1",
"Statement": [
{
"Effect": "Allow",
"Action": ["functiongraph:trigger:create"],
"Resource": ["urn:fss:*:*:function:*"]
}
]
}
}
}'Method 3: Assign to User
# Attach policy to user
hcloud iam v3 attach-user-policy \
--user-id "user-xxx" \
--policy-id "policy-xxx"Resource Scoping
All Functions in All Regions
"Resource": ["urn:fss:*:*:function:*"]Specific Region
"Resource": ["urn:fss:cn-north-4:*:function:*"]Specific Function
"Resource": [
"urn:fss:cn-north-4:project-id:function:default:my-function:*"
]Multiple Specific Functions
"Resource": [
"urn:fss:cn-north-4:project-id:function:default:backup-function:*",
"urn:fss:cn-north-4:project-id:function:default:cleanup-function:*"
]Condition Keys
Use conditions to further restrict access:
Time-based Restriction
{
"Effect": "Allow",
"Action": ["functiongraph:trigger:create"],
"Resource": ["*"],
"Condition": {
"DateLessThan": {
"g:CurrentTime": "2024-12-31T23:59:59Z"
}
}
}Source IP Restriction
{
"Effect": "Allow",
"Action": ["functiongraph:trigger:create"],
"Resource": ["*"],
"Condition": {
"IpAddress": {
"g:SourceIp": ["192.168.1.0/24", "10.0.0.0/8"]
}
}
}Role-Based Access Control (RBAC)
Built-in Roles
| Role | Description | Includes Trigger Permissions |
|---|---|---|
Tenant Administrator | Full access | Yes |
FunctionGraph Administrator | FunctionGraph full access | Yes |
FunctionGraph Developer | Create/manage functions | Yes |
FunctionGraph Viewer | Read-only access | No (view only) |
Custom Role for Automation
For CI/CD pipelines and automation:
{
"Version": "1.1",
"Statement": [
{
"Effect": "Allow",
"Action": [
"functiongraph:function:get",
"functiongraph:trigger:create",
"functiongraph:trigger:list"
],
"Resource": ["urn:fss:*:*:function:*"]
}
]
}Verifying Permissions
Check User Permissions
# List user's policies
hcloud iam v3 list-user-permissions --user-id "user-xxx"
# Check specific permission
hcloud functiongraph v2 list-functions --limit 1Permission Testing Script
import os
from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkfunctiongraph.v2 import FunctionGraphClient
def test_permission():
credentials = BasicCredentials(
ak=os.environ.get('HUAWEI_AK'),
sk=os.environ.get('HUAWEI_SK'),
project_id=os.environ.get('HUAWEI_PROJECT_ID')
)
client = FunctionGraphClient.new_builder() \
.with_credentials(credentials) \
.with_region(FunctionGraphRegion.value_of(os.environ.get('HUAWEI_REGION'))) \
.build()
try:
# Test list functions (requires functiongraph:function:list)
response = client.list_functions(limit=1)
print("✓ Has function list permission")
# Additional tests can be added here
except Exception as e:
print(f"✗ Permission denied: {e}")
test_permission()Permission Troubleshooting
Error: 403 Forbidden
Cause: Missing required IAM permission
Solution:
1. Check if functiongraph:trigger:create is in policy 2. Verify policy is attached to user/role 3. Check resource scope matches target function
Error: Unauthorized
Cause: Invalid or expired credentials
Solution:
1. Verify AK/SK are correct 2. Check if credentials have been rotated 3. Re-configure KooCLI authentication
Security Best Practices
1. Principle of Least Privilege: Grant only required permissions 2. Use Resource Restrictions: Limit to specific functions/regions 3. Add Conditions: Restrict by IP, time, or MFA 4. Separate Environments: Use different policies for dev/prod 5. Regular Audits: Review and remove unused permissions 6. Temporary Credentials: Use STS for short-lived access
Policy Examples by Use Case
Development Environment
{
"Version": "1.1",
"Statement": [
{
"Effect": "Allow",
"Action": [
"functiongraph:*"
],
"Resource": ["urn:fss:cn-north-4:*:function:dev-*:*"]
}
]
}Production Environment (Restricted)
{
"Version": "1.1",
"Statement": [
{
"Effect": "Allow",
"Action": [
"functiongraph:function:get",
"functiongraph:trigger:get",
"functiongraph:trigger:list"
],
"Resource": ["urn:fss:cn-north-4:*:function:prod-*:*"]
}
]
}CI/CD Pipeline
{
"Version": "1.1",
"Statement": [
{
"Effect": "Allow",
"Action": [
"functiongraph:function:get",
"functiongraph:trigger:create",
"functiongraph:trigger:update",
"functiongraph:trigger:delete"
],
"Resource": ["urn:fss:*:*:function:*"],
"Condition": {
"StringEquals": {
"g:UserName": "cicd-service-account"
}
}
}
]
}Related Documentation
Python SDK Installation Guide
Overview
This guide covers installation and configuration of the Huawei Cloud FunctionGraph Python SDK for creating and managing FunctionGraph triggers.
Prerequisites
- Python 3.9 or higher
- pip package manager
- Huawei Cloud account with Access Key (AK) and Secret Key (SK)
Installation
1. Install FunctionGraph SDK
# Install FunctionGraph SDK (includes core dependencies)
pip install huaweicloudsdkfunctiongraph
# Verify installation
python -c "from huaweicloudsdkfunctiongraph.v2 import FunctionGraphClient; print('SDK installed successfully')"2. Install Additional Dependencies (Optional)
# For enhanced functionality
pip install huaweicloudsdkcoreEnvironment Configuration
Configure authentication using environment variables:
Linux/macOS
export HUAWEI_AK="your_access_key"
export HUAWEI_SK="your_secret_key"
export HUAWEI_REGION="cn-north-4"
export HUAWEI_PROJECT_ID="your_project_id"Windows PowerShell
$env:HUAWEI_AK = "your_access_key"
$env:HUAWEI_SK = "your_secret_key"
$env:HUAWEI_REGION = "cn-north-4"
$env:HUAWEI_PROJECT_ID = "your_project_id"Windows Command Prompt
set HUAWEI_AK=your_access_key
set HUAWEI_SK=your_secret_key
set HUAWEI_REGION=cn-north-4
set HUAWEI_PROJECT_ID=your_project_idVerify Configuration
Create a test script to verify your configuration:
import os
from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkfunctiongraph.v2 import FunctionGraphClient
# Load credentials from environment
ak = os.environ.get("HUAWEI_AK")
sk = os.environ.get("HUAWEI_SK")
region = os.environ.get("HUAWEI_REGION")
project_id = os.environ.get("HUAWEI_PROJECT_ID")
# Create credentials
credentials = BasicCredentials(ak, sk, project_id)
# Create client
client = FunctionGraphClient.new_builder() \
.with_credentials(credentials) \
.with_region(region) \
.build()
print("Configuration verified successfully!")Common Regions
| Region Code | Region Name |
|---|---|
cn-north-4 | Beijing 4 |
cn-south-1 | Guangzhou |
cn-east-3 | Shanghai |
ap-southeast-1 | Hong Kong |
ap-southeast-3 | Singapore |
Virtual Environment Setup (Recommended)
Create Virtual Environment
# Create virtual environment
python -m venv venv
# Activate virtual environment
# Linux/macOS:
source venv/bin/activate
# Windows:
venv\Scripts\activateInstall SDK in Virtual Environment
pip install huaweicloudsdkfunctiongraphTroubleshooting
Issue 1: Module Not Found
# Ensure SDK is installed
pip install huaweicloudsdkfunctiongraph
# Verify Python path
python -c "import sys; print(sys.path)"Issue 2: Authentication Failed
# Verify environment variables are set
# Linux/macOS:
echo $HUAWEI_AK
echo $HUAWEI_SK
# Windows PowerShell:
$env:HUAWEI_AK
$env:HUAWEI_SKIssue 3: SSL Certificate Error
# Disable SSL verification (not recommended for production)
from huaweicloudsdkcore.http.http_config import HttpConfig
config = HttpConfig.get_default_http_config()
config.ignore_ssl_verification = True
client = FunctionGraphClient.new_builder() \
.with_http_config(config) \
.with_credentials(credentials) \
.with_region(region) \
.build()Issue 4: Region Not Found
Check that the region code is correct and matches your FunctionGraph function location.
Security Best Practices
1. Never commit credentials to version control 2. Use environment variables for all credential storage 3. Rotate AK/SK regularly 4. Use IAM temporary credentials when possible 5. Restrict IAM permissions to minimum required 6. Use virtual environments to isolate dependencies
SDK Version Management
# Check installed version
pip show huaweicloudsdkfunctiongraph
# Upgrade to latest version
pip install --upgrade huaweicloudsdkfunctiongraph
# List all Huawei Cloud SDKs
pip list | grep huaweicloudsdkUninstallation
# Uninstall SDK
pip uninstall huaweicloudsdkfunctiongraph
# Uninstall core SDK
pip uninstall huaweicloudsdkcoreNext Steps
After installation, proceed to:
- IAM Policies Configuration
- Verification Method
Verification Method
Overview
This document provides comprehensive methods to verify that FunctionGraph scheduled triggers have been correctly configured and are functioning as expected.
Verification Steps
Step 1: Confirm Trigger Creation
Using KooCLI
# List all triggers for the function
hcloud functiongraph v2 list-function-triggers \
--function-urn "urn:fss:cn-north-4:project-id:function:default:my-function:latest"Expected output structure:
{
"triggers": [
{
"trigger_id": "timer-xxx-xxx-xxx",
"trigger_type": "TIMER",
"trigger_name": "daily-trigger",
"trigger_config": "{\"name\":\"daily-trigger\",\"schedule\":\"0 0 2 * * ?\",\"scheduleType\":\"Rate\"}",
"enable_status": "active",
"created_at": "2024-01-15T10:00:00Z",
"updated_at": "2024-01-15T10:00:00Z"
}
]
}Using Python SDK
from huaweicloudsdkfunctiongraph.v2 import FunctionGraphClient
def verify_trigger(client, function_urn, trigger_name):
from huaweicloudsdkfunctiongraph.v2.model.list_function_triggers_request import ListFunctionTriggersRequest
request = ListFunctionTriggersRequest(function_urn=function_urn)
response = client.list_function_triggers(request)
for trigger in response.triggers:
if trigger.trigger_name == trigger_name:
print(f"✓ Trigger found: {trigger.trigger_id}")
print(f" Type: {trigger.trigger_type}")
print(f" Status: {trigger.enable_status}")
return trigger
print("✗ Trigger not found")
return NoneStep 2: Validate Trigger Configuration
Check Trigger Properties
# Query specific trigger type
hcloud functiongraph v2 show-function-trigger \
--function-urn "urn:fss:cn-north-4:project-id:function:default:my-function:latest" \
--trigger-type-codes "TIMER"Validation checklist:
| Property | Expected Value | Verification Method |
|---|---|---|
trigger_type | TIMER | Confirm in list output |
trigger_name | Matches input | Compare with specification |
enable_status | active | Verify in trigger details |
schedule | Valid Cron | Parse trigger_config JSON |
maxRetryTime | As configured | Check trigger_config |
Step 3: Validate Cron Expression
Cron Expression Parser
import json
from datetime import datetime
def validate_cron_config(trigger_config):
config = json.loads(trigger_config)
cron = config.get('schedule')
print(f"Cron Expression: {cron}")
# Parse and display schedule
parts = cron.split()
fields = ['Second', 'Minute', 'Hour', 'Day', 'Month', 'Weekday']
for field, value in zip(fields, parts):
print(f" {field}: {value}")
# Additional validation logic here
return TrueStep 4: Test Trigger Execution
Option A: Manual Trigger Invocation
# Invoke function manually to verify execution path
hcloud functiongraph v2 invoke-function \
--function-urn "urn:fss:cn-north-4:project-id:function:default:my-function:latest" \
--body '{"test": true}'Option B: Wait for Scheduled Execution
For triggers with long intervals, temporarily update to a shorter schedule:
# Temporarily set to 1-minute interval for testing
hcloud functiongraph v2 update-function-trigger \
--function-urn "..." \
--trigger-id "timer-xxx" \
--body '{
"trigger_config": "{\"schedule\":\"0 */1 * * * ?\",\"scheduleType\":\"Rate\"}"
}'
# Monitor for 1-2 executions
# Then restore original scheduleStep 5: Monitor Execution History
Via Console
1. Navigate to FunctionGraph Console 2. Select Functions → Click function name 3. Go to Triggers tab 4. Click Execution History for the trigger 5. Verify recent executions
Via API
def get_execution_history(client, function_urn, limit=10):
from huaweicloudsdkfunctiongraph.v2.model.list_function_statistics_request import ListFunctionStatisticsRequest
request = ListFunctionStatisticsRequest(
function_urn=function_urn,
period='1h' # Last hour
)
response = client.list_function_statistics(request)
print(f"Executions in last hour: {len(response.statistic)}")
for stat in response.statistic:
print(f" Duration: {stat.duration}ms")
print(f" Status: {stat.status}")Step 6: Verify Alarm Configuration
Ensure Cloud Eye alarms are configured:
# Check alarm rules for the function
hcloud ces v2 list-alarms \
--namespace "SYS.FunctionGraph" \
--dimensions "name=function_urn,value=urn:fss:..."Automated Verification Script
#!/usr/bin/env python3
import os
import json
from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkfunctiongraph.v2 import FunctionGraphClient, FunctionGraphRegion
class TriggerVerifier:
def __init__(self, ak, sk, project_id, region):
credentials = BasicCredentials(ak=ak, sk=sk, project_id=project_id)
self.client = FunctionGraphClient.new_builder() \
.with_credentials(credentials) \
.with_region(FunctionGraphRegion.value_of(region)) \
.build()
def verify(self, function_urn, trigger_name, expected_cron):
results = []
# 1. List triggers
triggers = self._list_triggers(function_urn)
trigger = next((t for t in triggers if t.trigger_name == trigger_name), None)
if not trigger:
return {"status": "failed", "error": "Trigger not found"}
# 2. Verify properties
results.append(("trigger_type", trigger.trigger_type == "TIMER"))
results.append(("enable_status", trigger.enable_status == "active"))
# 3. Verify cron
config = json.loads(trigger.trigger_config)
results.append(("cron_match", config.get("schedule") == expected_cron))
# 4. Summary
all_passed = all(r[1] for r in results)
return {
"status": "passed" if all_passed else "failed",
"trigger_id": trigger.trigger_id,
"checks": [{"name": n, "passed": p} for n, p in results]
}
def _list_triggers(self, function_urn):
from huaweicloudsdkfunctiongraph.v2.model.list_function_triggers_request import ListFunctionTriggersRequest
request = ListFunctionTriggersRequest(function_urn=function_urn)
response = self.client.list_function_triggers(request)
return response.triggers
# Usage
if __name__ == "__main__":
verifier = TriggerVerifier(
ak=os.environ.get("HUAWEI_AK"),
sk=os.environ.get("HUAWEI_SK"),
project_id=os.environ.get("HUAWEI_PROJECT_ID"),
region=os.environ.get("HUAWEI_REGION", "cn-north-4")
)
result = verifier.verify(
function_urn="urn:fss:cn-north-4:xxx:function:default:my-function:latest",
trigger_name="daily-trigger",
expected_cron="0 0 2 * * ?"
)
print(json.dumps(result, indent=2))Verification Checklist
Pre-Deployment
- [ ] Cron expression validated
- [ ] Function exists in target region
- [ ] IAM permissions confirmed
- [ ] Network connectivity verified
Post-Deployment
- [ ] Trigger appears in function trigger list
- [ ] Trigger type is
TIMER - [ ] Enable status is correct
- [ ] Cron expression matches specification
- [ ] Trigger ID returned from API
Post-Execution
- [ ] Function execution logs present
- [ ] No execution errors in history
- [ ] Execution duration within limits
- [ ] Memory usage acceptable
Troubleshooting Verification Failures
Trigger Not Found
Symptoms: Trigger not in list output
Checks:
1. Verify function URN is correct 2. Check if creation request succeeded 3. Verify trigger name spelling 4. Check region alignment
Trigger Not Executing
Symptoms: No execution history entries
Checks:
1. Confirm trigger status is active 2. Verify function is deployed (not failed state) 3. Check Cron expression validity 4. Review function timeout settings
Execution Failures
Symptoms: Execution history shows failures
Checks:
1. Review function code logs 2. Verify function has correct runtime 3. Check function timeout settings 4. Validate function input/output
Monitoring Commands
Real-time Monitoring
# Stream function logs
hcloud lts v2 list-logs \
--log-group-id "functiongraph-logs" \
--log-stream-id "my-function-logs" \
--start-time $(date -d '5 minutes ago' +%s)000Statistics Summary
# Get function statistics
hcloud functiongraph v2 list-function-statistics \
--function-urn "..." \
--period "1d"Related Documentation
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
FunctionGraph trigger creation tool
Create and configure TIMER triggers for FunctionGraph functions
"""
import os
import sys
import json
import logging
from typing import Dict, Any, Optional, Tuple
try:
from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkcore.exceptions.exceptions import ClientRequestException
from huaweicloudsdkfunctiongraph.v2.functiongraph_client import FunctionGraphClient
from huaweicloudsdkfunctiongraph.v2.region.functiongraph_region import FunctionGraphRegion
from huaweicloudsdkfunctiongraph.v2.model.create_function_trigger_request import CreateFunctionTriggerRequest
from huaweicloudsdkfunctiongraph.v2.model.create_function_trigger_request_body import CreateFunctionTriggerRequestBody
from huaweicloudsdkfunctiongraph.v2.model.trigger_event_data_request_body import TriggerEventDataRequestBody
except ImportError as e:
print(f"Please install SDK first: pip install huaweicloudsdkfunctiongraph (Error: {e})")
sys.exit(1)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class CronValidator:
"""Cron expression validator"""
FIELD_RANGES = {
'second': (0, 59),
'minute': (0, 59),
'hour': (0, 23),
'day': (1, 31),
'month': (1, 12),
'weekday': (1, 7)
}
@classmethod
def validate(cls, cron_expression: str) -> Tuple[bool, str]:
if not cron_expression:
return False, "Cron expression cannot be empty"
parts = cron_expression.strip().split()
if len(parts) < 6 or len(parts) > 7:
return False, f"Cron expression should have 6 or 7 fields, got {len(parts)}"
field_names = ['second', 'minute', 'hour', 'day', 'month', 'weekday']
for i, (part, field_name) in enumerate(zip(parts[:6], field_names)):
is_valid, error = cls._validate_field(part, field_name)
if not is_valid:
return False, f"Field '{field_name}' invalid: {error}"
return True, ""
@classmethod
def _validate_field(cls, field: str, field_name: str) -> Tuple[bool, str]:
min_val, max_val = cls.FIELD_RANGES[field_name]
if field == '*':
return True, ""
if field == '?':
if field_name in ['day', 'weekday']:
return True, ""
return False, "'?' can only be used for day or weekday fields"
if '-' in field:
try:
start, end = map(int, field.split('-'))
if start < min_val or end > max_val or start > end:
return False, f"Range should be between {min_val}-{max_val}"
return True, ""
except ValueError:
return False, "Invalid range format"
if field.startswith('*/'):
try:
step = int(field[2:])
if step < 1 or step > max_val:
return False, f"Step should be between 1-{max_val}"
return True, ""
except ValueError:
return False, "Invalid step format"
if ',' in field:
try:
values = [int(v) for v in field.split(',')]
for v in values:
if v < min_val or v > max_val:
return False, f"Value should be between {min_val}-{max_val}"
return True, ""
except ValueError:
return False, "Invalid list format"
try:
value = int(field)
if value < min_val or value > max_val:
return False, f"Value should be between {min_val}-{max_val}"
return True, ""
except ValueError:
return False, "Invalid numeric value"
class TriggerCreator:
"""Trigger creator for FunctionGraph"""
def __init__(self, ak: str, sk: str, region: str = 'cn-north-4', project_id: str = None):
self.ak = ak
self.sk = sk
self.region = region
self.project_id = project_id
self.client = self._init_client()
def _init_client(self) -> FunctionGraphClient:
credentials = BasicCredentials(ak=self.ak, sk=self.sk, project_id=self.project_id)
client = FunctionGraphClient.new_builder() \
.with_credentials(credentials) \
.with_region(FunctionGraphRegion.value_of(self.region)) \
.build()
return client
def validate_params(self, params: Dict[str, Any]) -> Tuple[bool, str]:
required_fields = ['function_urn', 'trigger_name', 'schedule']
for field in required_fields:
if not params.get(field):
return False, f"Missing required parameter: {field}"
schedule_type = params.get('schedule_type', 'Cron')
if schedule_type not in ['Rate', 'Cron']:
return False, "schedule_type must be 'Rate' or 'Cron'"
if schedule_type == 'Cron':
is_valid, error = CronValidator.validate(params['schedule'])
if not is_valid:
return False, f"Invalid Cron expression: {error}"
trigger_name = params['trigger_name']
if len(trigger_name) < 1 or len(trigger_name) > 64:
return False, "Trigger name must be 1-64 characters"
enable_status = params.get('enable_status', 'ACTIVE')
if enable_status not in ['ACTIVE', 'DISABLED']:
return False, "enable_status must be 'ACTIVE' or 'DISABLED'"
return True, ""
def check_function_exists(self, function_urn: str) -> Tuple[bool, str]:
try:
from huaweicloudsdkfunctiongraph.v2.model.show_function_config_request import ShowFunctionConfigRequest
request = ShowFunctionConfigRequest(function_urn=function_urn)
self.client.show_function_config(request)
return True, ""
except ClientRequestException as e:
if 'not found' in str(e).lower() or e.status_code == 404:
return False, "Target function not found"
return False, f"Failed to check function: {str(e)}"
def create_trigger(self, params: Dict[str, Any]) -> Dict[str, Any]:
is_valid, error_msg = self.validate_params(params)
if not is_valid:
return {'status': 'failed', 'error_code': 'InvalidParameter', 'message': error_msg}
exists, error = self.check_function_exists(params['function_urn'])
if not exists:
return {'status': 'failed', 'error_code': 'FunctionNotFound', 'message': error}
try:
event_data = TriggerEventDataRequestBody(
name=params['trigger_name'],
schedule_type=params.get('schedule_type', 'Cron'),
schedule=params['schedule']
)
user_event = params.get('user_event')
if user_event:
event_data.user_event = user_event
request_body = CreateFunctionTriggerRequestBody(
trigger_type_code='TIMER',
trigger_status=params.get('enable_status', 'ACTIVE'),
event_data=event_data
)
request = CreateFunctionTriggerRequest(
function_urn=params['function_urn'],
body=request_body
)
logger.info(f"Creating trigger: {params['trigger_name']}")
response = self.client.create_function_trigger(request)
trigger_name = params['trigger_name']
trigger_type_code = response.trigger_type_code or 'TIMER'
trigger_status = response.trigger_status or 'ACTIVE'
if response.event_data and hasattr(response.event_data, 'name') and response.event_data.name:
trigger_name = response.event_data.name
result = {
'status': 'success',
'trigger_id': response.trigger_id,
'trigger_name': trigger_name,
'trigger_type': trigger_type_code,
'schedule': params['schedule'],
'enable_status': trigger_status,
'message': 'Trigger created successfully'
}
logger.info(f"Trigger created successfully: {response.trigger_id}")
return result
except ClientRequestException as e:
error_code = e.error_code if hasattr(e, 'error_code') else 'Unknown'
error_msg = e.error_msg if hasattr(e, 'error_msg') else str(e)
logger.error(f"Failed to create trigger: {error_code} - {error_msg}")
if e.status_code == 409 or 'already exist' in error_msg.lower():
error_code = 'TriggerAlreadyExists'
elif 'limit' in error_msg.lower():
error_code = 'TriggerLimitExceeded'
return {'status': 'failed', 'error_code': error_code, 'message': error_msg}
except Exception as e:
logger.error(f"Exception creating trigger: {str(e)}")
return {'status': 'failed', 'error_code': 'InternalError', 'message': str(e)}
def load_config() -> Dict[str, str]:
config = {
'ak': os.environ.get('HUAWEI_AK'),
'sk': os.environ.get('HUAWEI_SK'),
'region': os.environ.get('HUAWEI_REGION', 'cn-north-4'),
'project_id': os.environ.get('HUAWEI_PROJECT_ID')
}
if not config['ak'] or not config['sk']:
raise ValueError("Please set environment variables HUAWEI_AK and HUAWEI_SK")
return config
def main():
import argparse
parser = argparse.ArgumentParser(description='Create FunctionGraph TIMER trigger')
parser.add_argument('--function-urn', required=True, help='Target function URN')
parser.add_argument('--name', required=True, help='Trigger name')
parser.add_argument('--schedule', required=True, help='Cron expression or Rate value')
parser.add_argument('--schedule-type', choices=['Cron', 'Rate'], default='Cron',
help='Schedule type: Cron (default) or Rate')
parser.add_argument('--status', choices=['ACTIVE', 'DISABLED'], default='ACTIVE',
help='Trigger status: ACTIVE (default) or DISABLED')
parser.add_argument('--user-event', default='', help='Additional user event data')
parser.add_argument('--skip-check', action='store_true',
help='Skip function existence check')
args = parser.parse_args()
try:
config = load_config()
except ValueError as e:
print(f"Config error: {e}")
sys.exit(1)
params = {
'function_urn': args.function_urn,
'trigger_name': args.name,
'schedule': args.schedule,
'schedule_type': args.schedule_type,
'enable_status': args.status,
'user_event': args.user_event
}
creator = TriggerCreator(
ak=config['ak'],
sk=config['sk'],
region=config['region'],
project_id=config['project_id']
)
if args.skip_check:
exists = True
else:
exists, error = creator.check_function_exists(params['function_urn'])
if not exists:
print(json.dumps({'status': 'failed', 'error_code': 'FunctionNotFound', 'message': error},
indent=2, ensure_ascii=False))
sys.exit(1)
result = creator.create_trigger(params)
print(json.dumps(result, indent=2, ensure_ascii=False))
if result['status'] != 'success':
sys.exit(1)
if __name__ == '__main__':
main()
def handler(event, context):
import json
from datetime import datetime
result = {
"message": "Timer triggered successfully",
"timestamp": datetime.utcnow().isoformat(),
"event": event
}
print(json.dumps(result, ensure_ascii=False))
return resultdef handler(event, context):
import json
from datetime import datetime
result = {
"message": "Timer triggered successfully",
"timestamp": datetime.utcnow().isoformat(),
"event": event
}
print(json.dumps(result, ensure_ascii=False))
return result