KEDAで始めるイベント駆動システム #k8snovice / keda-tutorial

イベント駆動システムを構築するツールとしてKEDAを調査したものです

Transcript

1. ,&%"Ͱ࢝ΊΔΠϕϯτۦಈγεςϜ ,VCFSOFUFT
   /PWJDF
   5PLZP
   
2. ͲͳͨͰ͔͢ʁ
   w $IJLBIJTB
   .JLBUBʢ5XJUUFS
   !NFMQP@NFMʣ
   w #UP#ͷ4BB4ձࣾʹॴଐ
   ͳʹͯ͠ΔਓͰ͔͢ʁ
   w όοΫΤϯυ։ൃ͕ϝΠϯɺ43&΋ͪΐΖͬͱ
   w ,VCFSOFUFTֶशதͷ΃ʔ΃ʔͰ͢ ࣗݾ঺հ
3. w ,&%"ͱ͸
   w ,&%"ͷػೳͱಛ௃
   w ,&%"ͷ࢖͍ํ
   w ·ͱΊ ຊ೔ͷൃද಺༰
4. ,&%"ͱ͸
5. ,&%"ͱ͸ ,VCFSOFUFT
   &WFOUESJWF
   "VUPTDBMJOHʢ,&%"ʣ͸Πϕϯτۦಈͷ ΦʔτεέʔϧΛ࣮ݱ͢ΔίϯϙʔωϯτͰ͢ɻ
   ೥ʹ$/$'ͷ4BOECPYϓϩδΣΫτͱͳ͓ͬͯΓɺݱࡏ͸ *ODVCBUJOHϓϩδΣΫτͱͳ͍ͬͯ·͢ɻ
6. ,&%"ͱ͸ ೚ҙͷ,VCFSOFUFTΫϥελʹ௥ՃͰ͖ΔܰྔίϯϙʔωϯτͰɺ )1"ͳͲඪ४ͷ,VCFSOFUFTίϯϙʔωϯτͱ࿈ܞͯ͠ಈ࡞͠·͢ɻ
   ΦʔτεέʔϧʹػೳΛߜ͍ͬͯΔͨΊγϯϓϧͳߏ଄Ͱɺগͳ͍Ϧ ιʔεͰಈ࡞ͤ͞Δ͜ͱ͕Ͱ͖·͢ɻ
7. ,&%"ͷػೳͱಛ௃
8. ,&%"ͷػೳͱಛ௃ w Πϕϯτͷঢ়ଶʹ߹Θͤͯ1PEΛεέʔϧͰ͖·͢ɻΠϕϯτ͕ൃੜ͍ͯ͠ͳ ͍৔߹͸1PE਺ΛʹͰ͖ΔͷϦιʔεΛ༗ޮʹ׆༻Ͱ͖·͢ɻ
   w Ωϡʔ΍%#ͳͲ༷ʑͳΠϕϯτϦιʔεΛτϦΨʔʹεέʔϧ͕ՄೳͰ͢ɻ ରԠ͍ͯ͠ΔΠϕϯτϦιʔε͸LBGLB
   1VMTBS
   .Z42-
   1PTUHSF42-ͳͲ ༷ʑͰ͢ɻৄࡉ͸ެࣜΛ֬ೝͯ͠Έ͍ͯͩ͘͞ɻ
   w ૊ΈࠐΈͷεέʔϥʔʢΠϕϯτϦιʔεΛ؂ࢹ͢Δ࢓૊Έʣͷ୅ΘΓʹಠࣗ ͷεέʔϥʔΛఆٛ͢Δ͜ͱ͕ՄೳͰ͢ɻ
9. w LFEBPQFSBUPS
    1PEΛ͔Β΁εέʔϧ͠·͢ɻ
   w LFEBPQFSBUPSNFUSJDTBQJTFSWFS
    ΠϕϯτσʔλΛ)1"ʹެ։͢Δ͜ͱ Ͱ1PEΛ͔ΒO΁εέʔϧ͠·͢ɻ
   w LFEBBENJTTJPOXFCIPPLT
    ϦιʔεͷมߋΛݕূ͠ߏ੒ϛεΛ๷ ͗·͢ɻ ΞʔΩςΫνϟ
10. ΧελϜϦιʔε w 4DBMFE0CKFDU
     %FQMPZNFOUͱ4UBUFGVM4FUΛεέʔϦϯά͠·͢ɻ
    w 4DBMFE+PC
     +PCΛεέʔϦϯά͠·͢ɻ
    w 5SJHHFS"VUIFOUJDBUJPOʢৄ͘͠৮Ε·ͤΜʣ
     ೝূ৘ใΛఆٛɺ4DMBFE0CKFDU͔ΒࢀরՄೳʹ͠·͢ɻ
    w $MVTUFS5SJHHFS"VUIFOUJDBUJPOʢৄ͘͠৮Ε·ͤΜʣ
     /NBFTQBDFΛލ͍Ͱར༻Մೳͳೝূ৘ใΛఆٛ͠·͢ɻ
11. Ϣʔεέʔε w 4DBMFE0CKFDU
     Ωϡʔͷ௕͞ʹ߹Θͤͯ1PEΛεέʔϧ͠·͢ɻΩϡʔ͕ۭʹͳΔ·Ͱ1PE ͸ॲཧΛଓߦ͠·͢ɻ
     ୹࣌ؒʹେྔͷϦΫΤετΛॲཧ͢ΔέʔεͰར༻͠·͢ɻ
    w 4DBMFE+PC
     ΩϡʔͷΠϕϯτʹରͯͭ͠ͷ1PEΛىಈ͠·͢ɻॲཧ͕૸Γ੾Δͱ 1PE͸ऴྃ͠·͢ɻ
     ௕࣌ؒͷॲཧΛ࣮ߦ͢ΔΑ͏ͳέʔεͰར༻͠·͢ɻ
12. ,&%"ͷ࢖͍ํ
13. ,&%" ,BGLB IUUQTHJUIVCDPN $POUBJOFS4PMVUJPOTLFEBLBGLB
    ͪ͜ΒͰެ։͞Ε͍ͯΔαϯϓϧΛར༻͠ ͯ؀ڥΛߏங͠·͢ɻ
    w 1SPEVDFSͷ1PE͸,BGLBͷΩϡʔʹϝ οηʔδΛొ࿥͠·͢ɻ
    w ,BGLBͷΩϡʔʹ߹Θͤͯ$POTVNFSͷ 1PEΛͭ·Ͱεέʔϧ͠·͢ɻ
    w ,BGLBͷΩϡʔ͕ۭʹͳΔͱ$POTVNFS ͷ1PE͸ʹεέʔϧ͠·͢ɻ ΰʔϧΠϝʔδ %FQMPZNFOU 1SPEVDFS $POTVNFS
14. ࣄલ४උ 
    4USJN[JΛར༻ͯ͠,BGLBΫϥελΛߏங͠·͢ɻ mikata: ~/ % kubectl get pods -n kafka NAME READY STATUS RESTARTS AGE kafdrop-5549954bdb-z24kv 1/1 Running 2 (24h ago) 7d18h my-cluster-entity-operator-5d8bccc5d4-8szq4 3/3 Running 6 (24h ago) 7d19h my-cluster-kafka-0 1/1 Running 2 (24h ago) 7d19h my-cluster-zookeeper-0 1/1 Running 2 (24h ago) 7d19h strimzi-cluster-operator-687b8c77db-8qx7w 1/1 Running 192 (13m ago) 8d mikata: ~/ % kubectl get svc -n kafka NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE kafdrop NodePort 10.96.39.41 <none> 9000:30080/TCP 7d18h my-cluster-kafka-bootstrap ClusterIP 10.96.62.244 <none> 9091/TCP,9092/TCP,9093/TCP 7d19h my-cluster-kafka-brokers ClusterIP None <none> 9090/TCP,9091/TCP,9092/TCP,9093/TCP 7d19h my-cluster-zookeeper-client ClusterIP 10.96.141.159 <none> 2181/TCP 7d19h my-cluster-zookeeper-nodes ClusterIP None <none> 2181/TCP,2888/TCP,3888/TCP 7d19h
15. ࣄલ४උ  ΧελϜϦιʔεͰ,BGLBͷ5PQJDΛ࡞੒͠·͢ɻ apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: labels: strimzi.io/cluster: my-cluster name: my-topic namespace: kafka spec: config: {} partitions: 5 replicas: 1 topicName: my-topic mikata: ~/ % kubectl get kafkatopic my-topic -n kafka NAME CLUSTER PARTITIONS REPLICATION FACTOR READY my-topic my-cluster 5 1 True
16. ࣄલ४උ  1SPEVDFSͱ$POTVNFSͷΞϓϦΛσϓϩΠ͠·͢ɻ mikata: ~/ % kubectl get po -n demo NAME READY STATUS RESTARTS AGE kafka-consumer-5468c48766-hnzl2 1/1 Running 0 6m27s kafka-producer-64696bbdd7-9b2xt 1/1 Running 0 6m42s mikata: ~/ % kubectl get svc -n demo NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE kafka-producer-service ClusterIP 10.96.193.144 <none> 80/TCP 7m34s ,BGLB 1SPEVDFS $POTVNFS
17. ࣄલ४උ  1SPEVDFS͕ϝοηʔδΛొ࿥ɺ$POTVNFS͸ϝοηʔδΛಡΈग़͠·͢ɻ ɾɾɾ for i := 0; i < messageCount; i++ { wg.Add(1) i := i go func() { defer wg.Done() writeMessage(w, i) }() } wg.Wait() ɾɾɾ } func writeMessage(w *kafka.Writer, i int) { message := kafka.Message{Key: []byte(fmt.Sprintf("Key- %d", i)), Value: []byte(fmt.Sprintf("Message No: %d, Time: %s", i, time.Now()))} err := w.WriteMessages(context.Background(), message) ɾɾɾ } ɾɾɾ r := kafka.NewReader(kafka.ReaderConfig{ Brokers: []string{kafkaBroker}, GroupID: consumerGroup, Topic: kafkaTopic, MinBytes: 10e1, // 1KB MaxBytes: 10e6, // 10MB }) defer r.Close() go func() { for { m, err := r.ReadMessage(context.Background()) if err != nil { break } time.Sleep(time.Second * time.Duration(sleepSeconds)) } }() }
18. 4DBMFE0CKFDUͷ࡞੒  )FMNΛར༻ͯ͠,&%"ΛΠϯετʔϧ͠·͢ɻ mikata: ~/ % helm repo add kedacore https://kedacore.github.io/charts mikata: ~/ % helm repo update mikata: ~/ % kubectl create ns keda mikata: ~/ % helm install keda kedacore -ns keda mikata: ~/ % kubectl get po -n keda NAME READY STATUS RESTARTS AGE keda-admission-webhooks-7df9d947c4-jhc4z 1/1 Running 1 (3d1h ago) 10d keda-operator-96579d64c-5tj9q 1/1 Running 3 (3d1h ago) 10d keda-operator-metrics-apiserver-8758c5874-xcmqd 1/1 Running 2 (3d1h ago) 10d mikata: ~/ % kubectl get svc -n keda NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE keda-admission-webhooks ClusterIP 10.96.129.95 <none> 443/TCP 10d keda-operator ClusterIP 10.96.152.106 <none> 9666/TCP 10d keda-operator-metrics-apiserver ClusterIP 10.96.98.38 <none> 443/TCP,80/TCP 10d
19. 4DBMFE0CKFDUͷ࡞੒ 
    ΧελϜϦιʔεͰ4DBMFE0CKFDUΛ࡞੒͠·͢ɻ apiVersion: keda.sh/v1alpha1 kind: ScaledObject metadata: name: kafka-consumer-scaler namespace: demo spec: scaleTargetRef: apiVersion: apps/v1 kind: Deployment name: kafka-consumer pollingInterval: 10 cooldownPeriod: 30 idleReplicaCount: 0 minReplicaCount: 1 maxReplicaCount: 5 triggers: - type: kafka metadata: bootstrapServers: my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092 consumerGroup: g1 topic: my-topic lagThreshold: "10" offsetResetPolicy: latest allowIdleConsumers: "false" version: 1.0.0 pollingInterval: τϦΨʔͷϙʔϦϯάִؒ cooldownPeriod: εέʔϧόοΫͷ଴ػظؒ idleReplicaCount: ଴ػ࣌ͷϨϓϦΧ਺ minReplicaCount: ࠷খϨϓϦΧ਺ maxReplicaCount: ࠷େϨϓϦΧ਺ bootstrapServers: kafkaͷ઀ଓઌ consumeGroup: Consumer Group໊ topic: topic໊ lagThreshold εέʔϧ͢Δᮢ஋
20. 4DBMFE0CKFDUͷ࡞੒  4DBMFE0CKFDUΛσϓϩΠ͠·͕͢ɺ1PE਺ʹมԽ͸͋Γ·ͤΜɻ mikata: ~/ % kubectl apply -f scaled-object.yaml scaledobject.keda.sh/kafka-consumer-scaler created mikata: ~/ % kubectl get scaledobjects -n demo NAME SCALETARGETKIND SCALETARGETNAME MIN MAX TRIGGERS AUTHENTICATION READY ACTIVE FALLBACK AGE kafka-consumer-scaler apps/v1.Deployment kafka-consumer 1 5 kafka True True False 3m23s mikata: ~/ % kubectl get po -n demo NAME READY STATUS RESTARTS AGE kafka-consumer-5468c48766-hnzl2 1/1 Running 0 2d19h kafka-producer-64696bbdd7-9b2xt 1/1 Running 0 2d20h mikata: ~/ % kubectl get hpa -n demo NAME REFERENCE TARGETS MINPODS MAXPODS REPLICAS AGE keda-hpa-kafka-consumer-scaler Deployment/kafka-consumer 5/10 (avg) 1 5 1 3m55s HPA͕࡞੒͞Ε·͢
21. εέʔϧͷ֬ೝ ΞϓϦʹෛՙΛ͔͚εέʔϧΛ֬ೝ͠·͢ɻ
    Ұ౓ͷϦΫΤετͰ1SPEVDFS͸ݸͷϝοηʔδΛLBGLBʹొ࿥͍ͯ͠·͢ɻ mikata: ~/ % kubectl port-forward svc/kafka-producer-service 8080:80 -n demo & [1] 10843 mikata: keda-kafka/ % Forwarding from 127.0.0.1:8080 -> 5550 Forwarding from [::1]:8080 -> 5550 mikata: ~/ % curl 127.0.0.1:8080 Handling connection for 8080 mikata: keda-kafka/ % kubectl get po -n demo NAME READY STATUS RESTARTS AGE kafka-consumer-5468c48766-6dvq7 1/1 Running 0 57s kafka-consumer-5468c48766-h4mfv 1/1 Running 0 42s kafka-consumer-5468c48766-h8vwt 1/1 Running 0 42s kafka-consumer-5468c48766-k4js6 1/1 Running 0 57s kafka-consumer-5468c48766-khqs6 1/1 Running 0 74s kafka-producer-64696bbdd7-9b2xt 1/1 Running 0 2d20h mikata: keda-kafka/ % kubectl get po -n demo NAME READY STATUS RESTARTS AGE kafka-producer-64696bbdd7-9b2xt 1/1 Running 0 2d20h
22. ·ͱΊ
23. ·ͱΊ Πϯετʔϧɺઃఆͷखॱ͕γϯϓϧͰಋೖͷোน͸௿͍ͷͰɺखͬऔΓૣ͘Π ϕϯτۦಈͷγεςϜΛߏங͢Δͷʹྑ͍ͱࢥ͍·͢ɻ
    खܰͳ൓໘ɺϝοηʔδϒϩʔΧʔͱͷ*'͸ఏڙ͞Εͳ͍ͷͰɺϒϩʔΧʔͱΞ ϓϦέʔγϣϯ͕ີ݁߹ʹͳΓ΍͍͢ͱײ͡·ͨ͠ɻ
    େن໛ͳ૊৫Ͱӡ༻͍ͯ͘͠৔߹͸ɺ͜ͷ͋ͨΓΛߟྀͭͭ͠ɺಋೖΛݕ౼͢ Δͷ͕ྑ͍ͷͰ͸ͳ͍͔ͱࢥ͍·͢ɻ
24. ͝ਗ਼ௌ͋Γ͕ͱ͏͍͟͝·ͨ͠