create_tidb_serverless_task.py 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657
  1. import time
  2. import click
  3. import app
  4. from configs import dify_config
  5. from core.rag.datasource.vdb.tidb_on_qdrant.tidb_service import TidbService
  6. from extensions.ext_database import db
  7. from models.dataset import TidbAuthBinding
  8. @app.celery.task(queue="dataset")
  9. def create_tidb_serverless_task():
  10. click.echo(click.style("Start create tidb serverless task.", fg="green"))
  11. tidb_serverless_number = dify_config.TIDB_SERVERLESS_NUMBER
  12. start_at = time.perf_counter()
  13. while True:
  14. try:
  15. # check the number of idle tidb serverless
  16. idle_tidb_serverless_number = TidbAuthBinding.query.filter(TidbAuthBinding.active == False).count()
  17. if idle_tidb_serverless_number >= tidb_serverless_number:
  18. break
  19. # create tidb serverless
  20. iterations_per_thread = 20
  21. create_clusters(iterations_per_thread)
  22. except Exception as e:
  23. click.echo(click.style(f"Error: {e}", fg="red"))
  24. break
  25. end_at = time.perf_counter()
  26. click.echo(click.style("Create tidb serverless task success latency: {}".format(end_at - start_at), fg="green"))
  27. def create_clusters(batch_size):
  28. try:
  29. new_clusters = TidbService.batch_create_tidb_serverless_cluster(
  30. batch_size,
  31. dify_config.TIDB_PROJECT_ID,
  32. dify_config.TIDB_API_URL,
  33. dify_config.TIDB_IAM_API_URL,
  34. dify_config.TIDB_PUBLIC_KEY,
  35. dify_config.TIDB_PRIVATE_KEY,
  36. dify_config.TIDB_REGION,
  37. )
  38. for new_cluster in new_clusters:
  39. tidb_auth_binding = TidbAuthBinding(
  40. cluster_id=new_cluster["cluster_id"],
  41. cluster_name=new_cluster["cluster_name"],
  42. account=new_cluster["account"],
  43. password=new_cluster["password"],
  44. )
  45. db.session.add(tidb_auth_binding)
  46. db.session.commit()
  47. except Exception as e:
  48. click.echo(click.style(f"Error: {e}", fg="red"))