Kore — ออกแบบ Message Broker + In-Memory Store ของตัวเอง

# Kore — ตัวอย่างการใช้งานจริง 12 ยูสเคส > เอกสารนี้เป็นภาคต่อของ `kore-design.md` ซึ่งอธิบายสถาปัตยกรรมและวิธีสร้าง > เอกสารฉบับนี้ตอบคำถามคนละข้อ คือ **"สร้างเสร็จแล้วเอาไปทำอะไรได้บ้าง"** > ทุกยูสเคสสมมติว่า Kore พัฒนาถึง Phase 5 แล้ว คือมี STREAM, TABLE, EPHEMERAL, > consumer group, WATCH, atomic batch, replication และ cluster ครบ --- ## สารบัญ **เริ่มต้น** 0. เครื่องมือที่ Kore ให้มา **กลุ่มที่ 1 — งาน integration และระบบองค์กร** 1. Idempotency และ rate limit ของ API Gateway 2. Event backbone แทนการต่อตรงแบบ point-to-point 3. เขียน state กับ event พร้อมกัน จนไม่ต้องมี outbox 4. สะพานเชื่อมระบบ legacy ที่ทำงานเป็นรอบ 5. Audit trail ที่ตรวจสอบย้อนหลังได้ 6. สถานะร่วมของ gateway หลายตัว **กลุ่มที่ 2 — งานแอปพลิเคชันทั่วไป** 7. Notification fan-out พร้อม retry และ dead letter 8. Session store และการเพิกถอน token ให้มีผลทันที 9. Job queue และงานตั้งเวลา 10. Materialized view สำหรับ dashboard เรียลไทม์ **กลุ่มที่ 3 — งานเฉพาะทาง** 11. Telemetry ingestion และการย่อข้อมูล 12. โหมด embedded สำหรับแอปเดียวจบ **สรุป** 13. ตารางสรุปว่าแต่ละยูสเคสใช้เครื่องมืออะไร 14. สิ่งที่ไม่ควรเอา Kore ไปทำ 15. เส้นทางย้ายจากระบบเดิม 16. ควรเริ่มจากยูสเคสไหนก่อน --- ## 0. เครื่องมือที่ Kore ให้มา ก่อนเข้ายูสเคส ต้องเห็นภาพว่ามีของอะไรให้ใช้บ้าง ทุกยูสเคสข้างล่างคือการผสมของ 8 ชิ้นนี้ ```mermaid flowchart TB subgraph PRIM["เครื่องมือพื้นฐานของ Kore"] ST["STREAM คือ log ที่เก็บตาม retention อ่านซ้ำได้ด้วย offset"] TB["TABLE คือ KV ที่ compact ต่อ key มี TTL และ index ใน memory"] EP["EPHEMERAL คือ pub sub ที่ไม่เขียนลงดิสก์"] CG["CONSUMER GROUP คือการแบ่ง partition ให้ผู้อ่านหลายตัว"] WT["WATCH คือการ subscribe การเปลี่ยนแปลงของ TABLE แบบ CDC"] AB["ATOMIC BATCH คือการเขียนหลาย record ข้าม namespace ให้ commit พร้อมกัน"] TL["TTL และ CAS สำหรับงานที่ต้องหมดอายุหรือแข่งกันเขียน"] EM["EMBEDDED MODE คือการฝัง Kore เป็น library ในแอป"] end N1>"ทุกอย่างวางอยู่บน log เส้นเดียวกัน จึงผสมกันในธุรกรรมเดียวได้ ซึ่งเป็นสิ่งที่ Kafka กับ Redis แยกกันทำไม่ได้"] PRIM --- N1 ``` | เครื่องมือ | เทียบได้กับของเดิม | ใช้เมื่อ | |---|---|---| | STREAM | Kafka topic | ต้องการลำดับ อ่านซ้ำ และให้ผู้อ่านหลายกลุ่มอ่านอิสระกัน | | TABLE | Redis keyspace | ต้องการค่าล่าสุดของ key แบบ random access เร็ว ๆ | | EPHEMERAL | Redis pub/sub | ส่งต่อทันที ยอมให้หายได้ เช่น presence หรือ typing indicator | | CONSUMER GROUP | Kafka consumer group | ต้องการขยายผู้ประมวลผลแบบขนานโดยไม่ทำงานซ้ำกัน | | WATCH | Debezium หรือ keyspace notification | ต้องการรู้ทันทีเมื่อค่าเปลี่ยน แบบเชื่อถือได้ | | ATOMIC BATCH | transactional outbox pattern | ต้องการให้ state กับ event เกิดพร้อมกันหรือไม่เกิดเลย | | TTL / CAS | Redis `SETEX` และ `SET NX` | ต้องการหมดอายุอัตโนมัติ หรือกันการเขียนชนกัน | | EMBEDDED | ไม่มีของเทียบใน Kafka | แอปเล็ก งาน edge หรือระบบที่ไม่อยากรันโปรเซสแยก | --- # กลุ่มที่ 1 — งาน integration และระบบองค์กร ## 1. Idempotency และ rate limit ของ API Gateway ### สถานการณ์ Gateway รันหลาย instance อยู่หลัง load balancer ปัญหาที่เจอทุกวันคือ - Client timeout แล้วยิงคำขอเดิมซ้ำ ทำให้เกิดรายการซ้ำที่ปลายทาง - Client บางรายยิงถี่จนกระทบรายอื่น - ทุก instance ต้องเห็นสถานะเดียวกัน จะเก็บใน memory ของตัวเองไม่ได้ ### เดิมทำอย่างไร ใช้ Redis เก็บ idempotency key และ token bucket ผลคือ **Redis กลายเป็น single point of failure ของ gateway ทั้งหมด** และถ้า Redis ตั้ง persistence เป็น async ข้อมูล idempotency อาจหายตอน failover ซึ่งแปลว่ารายการซ้ำหลุดไปได้จริง ### ทำด้วย Kore อย่างไร ใช้ TABLE สองตัวที่ตั้งค่าความทนทานคนละระดับ ซึ่งเป็นข้อได้เปรียบสำคัญ | Namespace | ชนิด | Durability | TTL | เหตุผล | |---|---|---|---|---| | `gw.idem` | TABLE | `quorum` | 24 ชั่วโมง | ห้ามหาย เพราะเป็นตัวกันรายการซ้ำที่เกี่ยวกับเงิน | | `gw.ratelimit` | TABLE | `none` | 60 วินาที | หายได้ อย่างมากคือปล่อยให้ยิงเกินไปนิดหน่อยชั่วครู่ | ```mermaid sequenceDiagram autonumber participant CL as Client participant GW as Gateway instance ใดก็ได้ participant KO as Kore participant BE as ระบบปลายทาง CL->>GW: ส่งคำขอพร้อม header idempotency key GW->>KO: INCR ที่ ratelimit ของ client รายนี้ KO-->>GW: ค่าปัจจุบันของหน้าต่างเวลานี้ alt เกินโควตา GW-->>CL: ตอบ 429 พร้อมบอกเวลาที่ให้ลองใหม่ else ยังไม่เกิน GW->>KO: SET แบบเขียนเฉพาะเมื่อยังไม่มี key นี้ alt มี key อยู่แล้ว KO-->>GW: ค่าเดิมที่เคยตอบไป GW-->>CL: ตอบผลเดิมทันที ไม่เรียกปลายทางซ้ำ else เป็นคำขอใหม่ KO-->>GW: จองสำเร็จ สถานะกำลังดำเนินการ GW->>BE: เรียกระบบปลายทาง BE-->>GW: ผลลัพธ์ GW->>KO: เขียนทับด้วยผลลัพธ์จริง GW-->>CL: ตอบผลลัพธ์ end end ``` ### โค้ดตัวอย่าง ```go // Reserve-then-fill pattern. The first writer wins and everyone else // gets the stored response instead of hitting the backend twice. func (g *Gateway) handle(ctx context.Context, idemKey string, req Request) (Response, error) { reserved, existing, err := g.kore.Table("gw.idem"). SetIfAbsent(ctx, idemKey, statusInFlight, 24*time.Hour) if err != nil { return Response{}, err } if !reserved { // Someone already handled or is handling this exact request. if existing.Status == statusInFlight { return Response{}, ErrDuplicateInFlight // tell the client to wait and retry } return existing.Response, nil } resp, err := g.backend.Call(ctx, req) if err != nil { // Release the reservation so a genuine retry can proceed. _ = g.kore.Table("gw.idem").Delete(ctx, idemKey) return Response{}, err } // Overwrite the reservation with the real answer. _ = g.kore.Table("gw.idem").Set(ctx, idemKey, resp, 24*time.Hour) return resp, nil } ``` ### ได้อะไร - ไม่ต้องรัน Redis แยก และไม่ต้องเลือกระหว่างเร็วกับปลอดภัย เพราะตั้งได้ต่อ namespace - idempotency key ที่ `quorum` จะไม่หายแม้ node ที่รับคำขอจะดับทันที - rate limit ที่ `none` เร็วเท่า Redis เพราะไม่แตะดิสก์เลย ### ต้องระวัง - **TTL ต้องยาวกว่าพฤติกรรม retry จริงของ client** ถ้า client retry ที่ 30 นาทีแต่ตั้ง TTL 10 นาที รายการจะซ้ำ เก็บสถิติจริงก่อนตั้งค่า - ต้องเก็บ **ผลลัพธ์** ไม่ใช่แค่ flag ว่าเคยทำแล้ว ไม่งั้น client ที่ retry จะไม่รู้ว่าผลเป็นอย่างไร - สถานะ `inFlight` ต้องมี TTL สั้นของตัวเอง กันกรณี gateway ดับกลางทางแล้ว key ค้าง --- ## 2. Event backbone แทนการต่อตรงแบบ point-to-point ### สถานการณ์ ระบบต้นทางหนึ่งตัว มีปลายทางที่ต้องรับข้อมูลเดียวกัน 5 ระบบ แต่ละระบบมีความเร็วและความเสถียรไม่เท่ากัน บางตัวปิดบำรุงรักษาทุกคืน ### เดิมทำอย่างไร ต้นทางเรียกปลายทางทีละตัว ผลที่ตามมา - เพิ่มปลายทางหนึ่งราย ต้องแก้และ deploy ต้นทางใหม่ - ปลายทางที่ช้าที่สุดกำหนดความเร็วของทั้งระบบ - ปลายทางตัวเดียวล่ม อาจทำให้ต้นทาง timeout ทั้งชุด - ปลายทางที่ล่มไปสองชั่วโมง ไม่มีทางได้ข้อมูลย้อนหลัง ```mermaid flowchart LR subgraph OLD["แบบเดิม ต่อตรงทุกคู่"] SRC1["ระบบต้นทาง"] --> D1["ปลายทาง 1"] SRC1 --> D2["ปลายทาง 2"] SRC1 --> D3["ปลายทาง 3 กำลังล่ม"] SRC1 --> D4["ปลายทาง 4 ช้ามาก"] SRC1 --> D5["ปลายทาง 5"] end subgraph NEW["แบบใหม่ ผ่าน Kore"] SRC2["ระบบต้นทาง"] -->|"เขียนครั้งเดียว"| STR["STREAM ชื่อ core.events"] STR --> G1["consumer group ของปลายทาง 1"] STR --> G2["consumer group ของปลายทาง 2"] STR --> G3["consumer group ของปลายทาง 3 ค้าง lag ไว้"] STR --> G4["consumer group ของปลายทาง 4 ตามช้าได้"] STR --> G5["consumer group ของปลายทาง 5"] end N2>"ต้นทางไม่รู้จักปลายทางอีกต่อไป เพิ่มปลายทางใหม่คือสร้าง consumer group ใหม่ ไม่ต้องแตะโค้ดต้นทางเลย"] NEW --- N2 ``` ### รายละเอียดการตั้งค่า | หัวข้อ | ค่าที่แนะนำ | เหตุผล | |---|---|---| | Partition key | รหัสของ entity หลัก เช่นเลขที่สัญญาหรือเลขบัญชี | ทำให้ event ของ entity เดียวกันเรียงกันเสมอ | | จำนวน partition | 2 เท่าของจำนวน consumer สูงสุดที่คาดว่าจะใช้ | เผื่อขยายโดยไม่ต้องเปลี่ยนการกระจาย key | | Retention | 7 ถึง 30 วัน | ให้ปลายทางที่ล่มนานกลับมาตามเก็บได้ | | Durability | `quorum` | เป็นข้อมูลหลักของระบบ | ### สิ่งที่ได้ทันทีจากการเปลี่ยนโครงสร้าง ```mermaid sequenceDiagram autonumber participant SR as ระบบต้นทาง participant KO as Kore STREAM participant C1 as ปลายทางปกติ participant C3 as ปลายทางที่ล่ม SR->>KO: append event เลข 1001 KO-->>SR: ack ภายในไม่กี่มิลลิวินาที KO->>C1: ส่งให้ทันที C1->>KO: commit offset 1001 Note over C3: ปลายทางนี้ปิดบำรุงรักษาสองชั่วโมง SR->>KO: append event ต่อไปอีกหลายพันรายการ KO-->>SR: ack ทุกครั้ง ต้นทางไม่รับรู้ว่ามีใครล่ม Note over C3: กลับมาออนไลน์ C3->>KO: ขอ offset ที่ commit ไว้ล่าสุด KO-->>C3: ส่ง event ที่ค้างทั้งหมดตามลำดับเดิม C3->>KO: commit offset ล่าสุด ``` ### ต้องระวัง - **Consumer ต้องทนต่อการได้รับข้อความซ้ำ** เพราะระบบรับประกันแบบ at-least-once ให้ทุก event มี `eventId` แล้วให้ปลายทางเก็บ dedupe ไว้ใน TABLE พร้อม TTL - ลำดับรับประกันแค่ภายใน partition ถ้าธุรกิจต้องการลำดับข้าม entity จะต้องออกแบบใหม่ ไม่ใช่แก้ด้วยการตั้งค่า - ตั้ง alert ที่ **consumer lag** ตั้งแต่วันแรก ไม่ใช่รอให้มีคนบ่น --- ## 3. เขียน state กับ event พร้อมกัน จนไม่ต้องมี outbox ### สถานการณ์ เมื่อสถานะของรายการเปลี่ยน ต้องทำสองอย่างพร้อมกัน คือบันทึกสถานะใหม่ และประกาศ event ให้ระบบอื่นรู้ **ทั้งสองอย่างต้องเกิดหรือไม่เกิดพร้อมกัน** ### ปัญหาที่เจอเมื่อแยกสองระบบ ```mermaid flowchart TB A["อัปเดตสถานะสำเร็จ"] --> B{"ส่ง event สำเร็จหรือไม่"} B -->|"สำเร็จ"| C["ปกติดี"] B -->|"ล้มเหลว"| D["สถานะเปลี่ยนแล้วแต่ไม่มีใครรู้ ระบบปลายน้ำไม่ตรงกันถาวร"] E["ส่ง event สำเร็จก่อน"] --> F{"อัปเดตสถานะสำเร็จหรือไม่"} F -->|"ล้มเหลว"| G["ประกาศไปแล้วว่าเกิดเหตุการณ์ ทั้งที่จริงยังไม่เกิด ปลายน้ำทำงานผิด"] N3>"ทางแก้เดิมคือ transactional outbox ซึ่งต้องมีตาราง outbox ตัว publisher แยก และงานทำความสะอาด รวมแล้วเป็นโค้ดหลายร้อยบรรทัดต่อหนึ่งระบบ"] D --- N3 ``` ### ทำด้วย Kore เขียนทั้งสองอย่างใน batch เดียว ซึ่ง commit ลง log เส้นเดียวกัน ```mermaid sequenceDiagram autonumber participant AP as Application participant KO as Kore participant CB as Commit Batcher participant CS as Consumer ปลายน้ำ AP->>KO: เปิด batch AP->>KO: ใส่คำสั่งอัปเดต TABLE ชื่อ loan.state AP->>KO: ใส่คำสั่ง append STREAM ชื่อ loan.events AP->>KO: สั่ง commit KO->>KO: append ทั้งสอง record ต่อกันใน log เส้นเดียวกัน KO->>CB: รอ durable ครั้งเดียวสำหรับทั้ง batch CB-->>KO: commit สำเร็จทั้งคู่ KO-->>AP: ตอบ ack ครั้งเดียว KO->>CS: ส่ง event ให้ปลายน้ำ Note over KO: ไม่มีสภาพที่ state เปลี่ยนแต่ event หาย หรือ event ออกไปแต่ state ไม่เปลี่ยน ``` ### โค้ดตัวอย่าง ```go // Both writes land in the same log commit: either both are durable or neither is. // This removes the need for an outbox table, a publisher worker, and a cleanup job. func (s *LoanService) Approve(ctx context.Context, loanID string, by string) error { state := LoanState{ID: loanID, Status: "APPROVED", ApprovedBy: by, At: time.Now()} event := LoanApproved{EventID: uuid.NewString(), LoanID: loanID, At: state.At} b := s.kore.Batch() b.Table("loan.state").Set(loanID, state, 0) // 0 means no expiry b.Stream("loan.events").Append(loanID, event) // partition key = loanID keeps order return b.Commit(ctx) // one fsync, one quorum round trip } ``` ### เทียบจำนวนงานที่ต้องทำ | สิ่งที่ต้องมี | แบบ outbox pattern | แบบ Kore | |---|---|---| | ตาราง outbox ในฐานข้อมูล | ต้องมี | ไม่ต้อง | | Worker คอยอ่าน outbox แล้วส่งออก | ต้องมี | ไม่ต้อง | | งานลบแถวเก่าออกจาก outbox | ต้องมี | ไม่ต้อง | | การจัดการกรณี worker ส่งซ้ำ | ต้องเขียนเอง | ระบบจัดการให้ | | ความหน่วงระหว่าง commit กับ event ออก | หลักวินาทีตามรอบ polling | ทันที | | จำนวนโค้ดโดยประมาณ | หลายร้อยบรรทัดต่อหนึ่งบริการ | 4 บรรทัด | ### ต้องระวัง - Atomic batch ทำงานได้เมื่อทุก record อยู่ใน **partition ที่ Raft group เดียวกัน** ถ้าข้าม group จะต้องใช้ transaction ข้าม group ซึ่งแพงกว่ามาก วิธีเลี่ยงคือใช้ **key เดียวกัน** เป็น partition key ของทั้ง TABLE และ STREAM ซึ่งเป็นสิ่งที่ควรทำอยู่แล้วโดยธรรมชาติ - ขนาดของ batch ควรจำกัด ไม่ควรยัดหลายพัน record ในคำขอเดียวเพราะจะไปหน่วงคำขออื่นในคิว --- ## 4. สะพานเชื่อมระบบ legacy ที่ทำงานเป็นรอบ ### สถานการณ์ ระบบหลักส่งข้อมูลออกมาเป็นไฟล์ทุก 15 นาที หรือให้ดึงผ่าน stored procedure เป็นรอบ แต่ระบบใหม่ต้องการข้อมูลแบบต่อเนื่อง และต้องการย้อนดูได้ว่าเมื่อวานเวลานั้นข้อมูลหน้าตาเป็นอย่างไร ### ปัญหาคลาสสิกของ connector แบบนี้ connector ดับตอนอ่านไฟล์ไปครึ่งหนึ่ง พอ restart แล้วอ่านซ้ำทั้งไฟล์ ทำให้ข้อมูลซ้ำ หรือถ้าจดว่าอ่านถึงไหนไว้ในไฟล์แยก ก็ไม่มีอะไรรับประกันว่าการจดกับการส่งจะสำเร็จพร้อมกัน ### ทำด้วย Kore ```mermaid flowchart TB A["connector ตรวจพบไฟล์ใหม่"] --> B["อ่าน checkpoint ของไฟล์นี้จาก TABLE"] B --> C{"เคยประมวลผลจบแล้วหรือยัง"} C -->|"จบแล้ว"| D["ข้ามไฟล์นี้"] C -->|"ยังไม่จบ"| E["เปิดไฟล์แล้ว seek ไปที่บรรทัดที่ค้างไว้"] E --> F["อ่านทีละก้อน เช่นก้อนละ 500 บรรทัด"] F --> G["เปิด batch"] G --> H["append 500 record เข้า STREAM"] H --> I["อัปเดต checkpoint ของไฟล์ใน TABLE ใน batch เดียวกัน"] I --> J["commit"] J --> K{"หมดไฟล์แล้วหรือยัง"} K -->|"ยังไม่หมด"| F K -->|"หมดแล้ว"| L["ทำเครื่องหมายว่าไฟล์นี้จบสมบูรณ์"] L --> M["ย้ายไฟล์ไปโฟลเดอร์เก็บถาวร"] N4>"เพราะ checkpoint กับข้อมูลอยู่ใน commit เดียวกัน connector จึงดับตอนไหนก็ได้ พอกลับมาจะอ่านต่อจากจุดเดิมพอดี ไม่ซ้ำและไม่ขาด"] J --- N4 ``` ### โค้ดตัวอย่าง ```go // The checkpoint and the data are committed together, which makes the // connector crash-safe without any external coordination. func (c *FileConnector) processChunk(ctx context.Context, f *FileState, rows []Row) error { b := c.kore.Batch() for _, r := range rows { b.Stream("host.txn").Append(r.AccountNo, r) } f.LastLine += len(rows) b.Table("connector.checkpoint").Set(f.Name, f, 30*24*time.Hour) return b.Commit(ctx) } ``` ### ได้อะไร - ระบบใหม่เห็นข้อมูลเป็น stream ต่อเนื่อง ทั้งที่ต้นทางยังทำงานเป็นรอบเหมือนเดิม ไม่ต้องแตะระบบหลักเลย - ย้อนเวลาได้ เพราะ STREAM เก็บทุก record ตามลำดับพร้อม timestamp - เพิ่มระบบที่อยากใช้ข้อมูลชุดนี้ได้เรื่อย ๆ โดยไม่ต้องขอสิทธิ์เข้าถึงระบบหลักเพิ่ม ซึ่งมักเป็นขั้นตอนที่ใช้เวลานานที่สุดในองค์กร ### ต้องระวัง - ไฟล์ที่ต้นทางเขียนยังไม่เสร็จ ต้องรอสัญญาณว่าเขียนจบ เช่นไฟล์ `.done` คู่กัน หรือดูจากการที่ขนาดไฟล์นิ่งแล้ว - ถ้าต้นทางส่งไฟล์เดิมซ้ำ ให้ใช้ hash ของเนื้อไฟล์เป็นส่วนหนึ่งของ checkpoint key - ข้อมูลจากระบบ legacy มักมีรหัสตัวอักษรและรูปแบบวันที่แปลก ๆ ให้แปลงที่ connector ชั้นเดียว แล้ว **เก็บของดิบไว้ใน STREAM แยกอีกเส้นด้วย** เผื่อวันหนึ่งพบว่าแปลงผิด จะได้ประมวลผลใหม่จากของดิบได้ --- ## 5. Audit trail ที่ตรวจสอบย้อนหลังได้ ### สถานการณ์ ระบบที่มีการกำกับดูแลต้องพิสูจน์ได้ว่า ใครทำอะไร เมื่อไร และหลักฐานนั้น **ไม่ถูกแก้ไขย้อนหลัง** ปัญหาของการเก็บ audit ลงตารางในฐานข้อมูลคือคนที่มีสิทธิ์ระดับ DBA แก้ได้ และไม่มีร่องรอย ### ทำด้วย Kore STREAM เป็น append-only โดยธรรมชาติอยู่แล้ว เพิ่มการผูก hash ต่อกันเข้าไปจะได้ **tamper-evident chain** คือแก้ได้แต่จะรู้ทันที ```mermaid flowchart LR R1["record 1 มี prevHash เป็นค่าเริ่มต้น"] --> R2["record 2 มี prevHash เท่ากับ hash ของ record 1"] R2 --> R3["record 3 มี prevHash เท่ากับ hash ของ record 2"] R3 --> R4["record 4 มี prevHash เท่ากับ hash ของ record 3"] V["ตัวตรวจสอบเดินไล่ทั้งสาย"] --> R1 V --> CHK{"hash ต่อกันครบทุกข้อหรือไม่"} CHK -->|"ครบ"| OK["ยืนยันได้ว่าไม่มีการแก้ไข"] CHK -->|"ขาดตอน"| BAD["ระบุได้ว่าถูกแก้ที่ record ไหน"] N5>"เก็บ hash ของ record ล่าสุดไว้นอกระบบเป็นระยะ เช่นส่งเข้าอีเมลหรือระบบ log กลาง จะทำให้แก้ทั้งสายย้อนหลังไม่ได้"] R4 --- N5 ``` ### โครงสร้างที่ใช้ | Namespace | ชนิด | ตั้งค่า | หน้าที่ | |---|---|---|---| | `audit.log` | STREAM | retention 7 ปี ปิด compaction | เก็บเหตุการณ์ทั้งหมดตามลำดับ | | `audit.chain` | TABLE | durability `quorum` | เก็บ hash ล่าสุดของแต่ละ partition | | `audit.byref` | TABLE | durability `quorum` | index จากเลขอ้างอิงไปยัง offset เพื่อค้นเร็ว | | `audit.key` | TABLE | durability `quorum` | เก็บกุญแจเข้ารหัสรายบุคคล สำหรับกรณีต้องลบข้อมูล | ### ประเด็นที่คนมักคิดไม่ถึง คือกฎหมายให้ลบข้อมูลส่วนบุคคล Append-only ขัดกับสิทธิในการขอให้ลบข้อมูลโดยตรง ทางแก้ที่ใช้ได้จริงคือ **crypto shredding** ```mermaid sequenceDiagram autonumber participant AP as Application participant KO as Kore participant AU as ผู้ตรวจสอบ AP->>KO: เขียน audit record โดยเข้ารหัสส่วนที่เป็นข้อมูลส่วนบุคคลด้วยกุญแจเฉพาะราย AP->>KO: เก็บกุญแจไว้ใน TABLE ชื่อ audit.key Note over KO: ส่วนที่ไม่ใช่ข้อมูลส่วนบุคคล เช่นเวลาและชนิดของเหตุการณ์ ไม่ได้เข้ารหัส AU->>KO: ขอตรวจสอบย้อนหลัง KO-->>AU: อ่านได้ทั้งหมดเพราะมีกุญแจอยู่ AP->>KO: เมื่อมีคำขอให้ลบข้อมูล ให้ลบเฉพาะกุญแจของรายนั้น Note over KO: record ยังอยู่ chain ยังตรวจสอบได้ แต่เนื้อหาส่วนบุคคลถอดรหัสไม่ได้อีกต่อไป ``` ### ต้องระวัง - **เก็บ hash ล่าสุดไว้นอกระบบด้วย** ไม่งั้นคนที่ควบคุมทั้งระบบสามารถสร้าง chain ใหม่ทั้งสายได้ - Retention 7 ปีหมายถึงข้อมูลจะโตมาก ควรตั้ง compression และวางแผนย้าย segment เก่าไปเก็บที่ถูกกว่าตั้งแต่แรก - อย่าเก็บข้อมูลส่วนบุคคลลง STREAM แบบไม่เข้ารหัส เพราะลบทีหลังไม่ได้จริง --- ## 6. สถานะร่วมของ gateway หลายตัว ### สถานการณ์ Gateway หลาย instance ต้องเห็นภาพเดียวกันว่า - ปลายทางตัวไหนกำลังล่ม จะได้ไม่ยิงไปซ้ำเติม - ค่า config เช่น timeout หรือ routing เปลี่ยนแล้วต้องมีผลทันทีโดยไม่ต้อง restart - โควตาที่แชร์กันใช้ไปเท่าไรแล้ว ### ปัญหาของวิธีเดิม Redis pub/sub เป็น fire-and-forget instance ที่ขาดการเชื่อมต่อไปครู่หนึ่งจะพลาดข้อความ แล้วจะไม่มีทางรู้ว่าตัวเองพลาด ผลคือ gateway บางตัวใช้ config เก่าอยู่เป็นชั่วโมงโดยไม่มีใครสังเกต ### ทำด้วย Kore ด้วย WATCH WATCH ไม่ใช่ pub/sub แต่เป็นการ subscribe การเปลี่ยนแปลงของ TABLE ที่ **มี offset กำกับ** จึงรู้ได้ว่าตัวเองพลาดช่วงไหน และตามเก็บได้ ```mermaid sequenceDiagram autonumber participant OP as ผู้ดูแลระบบ participant KO as Kore TABLE ชื่อ gw.config participant G1 as Gateway instance 1 participant G2 as Gateway instance 2 G1->>KO: WATCH ตั้งแต่ offset ล่าสุดที่เคยเห็น G2->>KO: WATCH ตั้งแต่ offset ล่าสุดที่เคยเห็น OP->>KO: แก้ค่า timeout ของปลายทางหนึ่ง KO->>G1: push การเปลี่ยนแปลงพร้อม offset KO->>G2: push การเปลี่ยนแปลงพร้อม offset G1->>G1: อัปเดต cache ใน memory ของตัวเอง Note over G2: instance นี้เพิ่งขาดการเชื่อมต่อไปสามสิบวินาที G2->>KO: เชื่อมต่อใหม่ พร้อมแจ้ง offset สุดท้ายที่ได้รับ KO-->>G2: ส่งการเปลี่ยนแปลงที่พลาดไปทั้งหมดตามลำดับ G2->>G2: อัปเดต cache จนตามทัน ``` ### โครงสร้างที่ใช้ | Namespace | ชนิด | ตั้งค่า | หน้าที่ | |---|---|---|---| | `gw.config` | TABLE | `quorum` ไม่มี TTL | ค่าตั้งค่าที่ต้องมีผลทันที | | `gw.health` | TABLE | `none` TTL 15 วินาที | สถานะปลายทาง ถ้าไม่มีใครเขียนต่อจะหายเอง | | `gw.breaker` | TABLE | `async` TTL 60 วินาที | สถานะ circuit breaker ที่แชร์กัน | **เทคนิคที่ใช้ได้ดี** ตั้ง TTL ให้ `gw.health` สั้นกว่ารอบการเขียนสองเท่า ถ้า instance ที่คอยเช็คสุขภาพตาย ค่าจะหมดอายุไปเอง ไม่ต้องมีใครมาคอยลบ ซึ่งเป็นการใช้ TTL เป็นกลไก failure detection ในตัว ### ต้องระวัง - Circuit breaker ที่แชร์กันมีผลข้างเคียง คือถ้า instance เดียวเห็น error แล้วเปิด breaker จะกระทบทุก instance ควรใช้เกณฑ์แบบนับจากหลาย instance ไม่ใช่เปิดทันทีจากรายเดียว - WATCH ทำให้ทุก instance ได้รับข้อความทุกครั้งที่ค่าเปลี่ยน ถ้ามี key ที่เปลี่ยนถี่มากอย่าไว้ใน namespace เดียวกับ config ที่นาน ๆ เปลี่ยนที --- # กลุ่มที่ 2 — งานแอปพลิเคชันทั่วไป ## 7. Notification fan-out พร้อม retry และ dead letter ### สถานการณ์ เหตุการณ์หนึ่งต้องแจ้งเตือนผ่านหลายช่องทาง เช่น push, อีเมล, แชท ปลายทางแต่ละช่องล่มไม่พร้อมกัน และมีข้อจำกัดอัตราการส่งไม่เท่ากัน ### ออกแบบด้วย Kore ```mermaid flowchart TB EV["event ต้นทางเข้ามาที่ STREAM ชื่อ notify.requested"] --> RT["consumer อ่านแล้วดูค่าตั้งค่าผู้ใช้จาก TABLE"] RT --> DD{"เคยส่งเรื่องนี้ให้คนนี้แล้วหรือยัง"} DD -->|"เคยแล้ว"| SK["ข้าม เพื่อกันการส่งซ้ำเมื่อ consumer อ่านข้อความซ้ำ"] DD -->|"ยังไม่เคย"| FO["แตกออกเป็นงานย่อยตามช่องทางที่ผู้ใช้เปิดไว้"] FO --> Q1["STREAM ชื่อ notify.push"] FO --> Q2["STREAM ชื่อ notify.email"] FO --> Q3["STREAM ชื่อ notify.chat"] Q1 --> W1["worker ของช่องทาง push"] Q2 --> W2["worker ของช่องทาง email"] Q3 --> W3["worker ของช่องทาง chat"] W1 --> RS{"ส่งสำเร็จหรือไม่"} W2 --> RS W3 --> RS RS -->|"สำเร็จ"| OKD["บันทึกผลลง TABLE ชื่อ notify.result"] RS -->|"ล้มเหลวแบบชั่วคราว"| RTY{"ครบจำนวนครั้งที่ยอมให้ลองหรือยัง"} RS -->|"ล้มเหลวถาวร เช่นเลขปลายทางผิด"| DLQ["ส่งเข้า STREAM ชื่อ notify.dlq"] RTY -->|"ยังไม่ครบ"| DLY["เขียนเข้า TABLE ชื่อ notify.delayed พร้อม TTL ตามระยะถอย"] RTY -->|"ครบแล้ว"| DLQ DLY -->|"เมื่อ TTL หมด ระบบจะส่งสัญญาณผ่าน WATCH"| Q1 N6>"ใช้ TTL ที่หมดอายุเป็นตัวจับเวลา retry ทำให้ไม่ต้องมี scheduler แยก และงานที่ตั้งเวลาไว้จะไม่หายแม้ระบบรีสตาร์ท"] DLY --- N6 ``` ### ตารางระยะถอยที่แนะนำ | ครั้งที่ | รอก่อนลองใหม่ | เหตุผล | |---|---|---| | 1 | 10 วินาที | ปัญหาเครือข่ายชั่วคราวมักหายในช่วงนี้ | | 2 | 1 นาที | ปลายทางที่รีสตาร์ทมักกลับมาในช่วงนี้ | | 3 | 5 นาที | | | 4 | 30 นาที | | | 5 | 2 ชั่วโมง | หลังจากนี้ให้เข้า dead letter และแจ้งคน | เพิ่ม jitter สุ่มบวกลบ 20% ทุกครั้ง เพื่อไม่ให้งานที่ล้มเหลวพร้อมกันกลับมาชนกันอีกรอบ ### ต้องระวัง - **Dead letter ที่ไม่มีคนดูเท่ากับไม่มี** ต้องมี dashboard และ alert เมื่อมีของเข้า และต้องมีคำสั่งให้ส่งกลับเข้าคิวปกติได้หลังแก้ปัญหาแล้ว - แยก error ชั่วคราวกับถาวรให้ชัด การ retry เลขปลายทางที่ผิดรูปแบบ 5 ครั้งคือการเสียเวลาเปล่า - เก็บผลการส่งไว้ใน TABLE พร้อม TTL เพื่อตอบคำถาม "ทำไมลูกค้าไม่ได้รับ" ได้ภายในไม่กี่วินาที --- ## 8. Session store และการเพิกถอน token ให้มีผลทันที ### สถานการณ์ ระบบใช้ JWT ที่มีอายุ 1 ชั่วโมง เพื่อไม่ต้องเช็คฐานข้อมูลทุกคำขอ แต่เมื่อผู้ใช้กดออกจากระบบทุกอุปกรณ์ หรือฝ่ายความปลอดภัยสั่งระงับบัญชี **token ที่ออกไปแล้วต้องใช้ไม่ได้ทันที** ไม่ใช่รออีก 59 นาที ### ปัญหาของวิธีเดิม เก็บ blacklist ใน Redis แล้วให้ทุกคำขอไปเช็ค ทำให้ Redis กลายเป็นคอขวดและเป็นจุดล้มเหลวเดี่ยว ถ้าทำ local cache แทน ก็ไม่รู้ว่าเมื่อไรควรล้าง cache ### ทำด้วย Kore ```mermaid sequenceDiagram autonumber participant SEC as ระบบความปลอดภัย participant KO as Kore participant A1 as App instance 1 participant A2 as App instance 2 participant U as ผู้ใช้ A1->>KO: WATCH namespace ชื่อ auth.revoked A2->>KO: WATCH namespace ชื่อ auth.revoked SEC->>KO: SET รหัส token ที่ถูกเพิกถอน พร้อม TTL เท่ากับอายุที่เหลือของ token KO->>A1: push ทันที KO->>A2: push ทันที A1->>A1: ใส่ลงชุด revoked ใน memory ของตัวเอง A2->>A2: ใส่ลงชุด revoked ใน memory ของตัวเอง U->>A2: ส่งคำขอพร้อม token ที่ถูกเพิกถอน A2->>A2: ตรวจจาก memory ล้วน ไม่ต้องเรียก Kore A2-->>U: ปฏิเสธด้วย 401 Note over KO: เมื่อ token หมดอายุตามธรรมชาติ TTL จะลบรายการทิ้งเอง ชุด revoked จึงไม่โตไม่มีที่สิ้นสุด ``` **จุดที่ฉลาดคือตั้ง TTL เท่ากับอายุที่เหลือของ token พอดี** เพราะหลังจากนั้น token ก็หมดอายุเองอยู่แล้ว ไม่ต้องจำต่อ ทำให้ชุด blacklist มีขนาดคงที่ตามจำนวนการเพิกถอนต่อชั่วโมง ไม่ใช่สะสมไปเรื่อย ๆ ### โครงสร้างที่ใช้ | Namespace | ชนิด | ตั้งค่า | หน้าที่ | |---|---|---|---| | `auth.session` | TABLE | `async` TTL แบบต่ออายุเมื่อมีการใช้งาน | ข้อมูล session ที่ต้องอ่านบ่อย | | `auth.revoked` | TABLE | `quorum` TTL เท่าอายุที่เหลือของ token | รายการเพิกถอน ห้ามหาย | | `auth.events` | STREAM | retention 90 วัน | ประวัติการเข้าออกระบบสำหรับตรวจสอบ | ### ต้องระวัง - ตอนแอป start ต้องโหลดชุด revoked ทั้งหมดก่อนเริ่มรับ traffic ไม่งั้นช่วงแรกจะปล่อย token ที่ถูกเพิกถอนผ่าน - ถ้าเชื่อมต่อ WATCH ขาดนานเกินเกณฑ์ ควรปฏิเสธคำขอหรือเปลี่ยนไปเช็คแบบ online ชั่วคราว ดีกว่าปล่อยผ่านโดยใช้ข้อมูลเก่า - นาฬิกาของแต่ละเครื่องต้องตรงกันพอสมควร เพราะ TTL ผูกกับเวลา --- ## 9. Job queue และงานตั้งเวลา ### สถานการณ์ ต้องการคิวงานที่ทำงานขนานได้ มี worker หลายตัว งานที่ทำไม่สำเร็จต้องกลับมาใหม่ และต้องมีงานตั้งเวลาล่วงหน้าได้ เช่นส่งใบแจ้งหนี้ในอีก 3 วัน ### ออกแบบด้วย Kore ```mermaid flowchart TB subgraph SUB["งานที่ทำทันที"] P["ผู้สร้างงาน append เข้า STREAM ชื่อ jobs.ready"] --> CG["consumer group ของ worker"] CG --> W1["worker 1"] CG --> W2["worker 2"] CG --> W3["worker 3"] end subgraph LEASE["การจองงานและตรวจว่า worker ยังอยู่"] W1 --> LS["เขียน lease ลง TABLE ชื่อ jobs.lease พร้อม TTL 30 วินาที"] LS --> HB["worker ต่ออายุ lease ทุก 10 วินาทีระหว่างทำงาน"] HB --> FIN{"ทำงานเสร็จหรือไม่"} FIN -->|"เสร็จ"| CM["commit offset แล้วลบ lease"] FIN -->|"worker ดับ ไม่มีใครต่ออายุ"| EXP["lease หมดอายุเอง งานถูกหยิบไปทำใหม่"] end subgraph DELAY["งานตั้งเวลา"] SC["เขียนงานลง TABLE ชื่อ jobs.scheduled พร้อม TTL เท่ากับเวลาที่ต้องรอ"] --> TT["เมื่อ TTL หมด ระบบจะสร้าง event หมดอายุ"] TT --> MV["consumer ที่ฟัง event หมดอายุ ย้ายงานเข้า jobs.ready"] end N7>"ใช้ TTL ทำหน้าที่ทั้งจับเวลางานล่วงหน้า และตรวจจับ worker ที่ตายไปเงียบ ๆ จึงไม่ต้องมี scheduler หรือ heartbeat monitor แยก"] LEASE --- N7 ``` ### ตารางเทียบกับวิธีเดิม | ความต้องการ | แบบเดิม | แบบ Kore | |---|---|---| | คิวงานขนาน | Redis list หรือ RabbitMQ | STREAM กับ consumer group | | งานตั้งเวลา | cron หรือ Redis sorted set บวก worker คอย poll | TTL ที่หมดอายุแล้วสร้าง event | | ตรวจว่า worker ตาย | heartbeat monitor แยก | lease ที่มี TTL ต่ออายุ | | ประวัติงานที่ทำไปแล้ว | ต้องเก็บลงฐานข้อมูลเอง | มีอยู่แล้วใน STREAM | | ดูว่ามีงานค้างเท่าไร | ต้องนับเอง | ดูจาก consumer lag ได้ตรง ๆ | ### ต้องระวัง - **ลำดับกับความขนานขัดกันเสมอ** ถ้าอยากให้งานของลูกค้ารายเดียวกันทำตามลำดับ ให้ใช้รหัสลูกค้าเป็น partition key แล้วยอมรับว่าลูกค้ารายที่มีงานเยอะจะเป็นคอขวดของ partition นั้น - งานที่ใช้เวลานานกว่า lease ต้องต่ออายุระหว่างทำ ไม่ใช่ตั้ง lease ยาว ๆ ไว้ก่อน เพราะถ้า worker ตายจริงจะต้องรอนานเกินจำเป็น - อย่าเก็บ payload ก้อนใหญ่ไว้ในงาน ให้เก็บแค่รหัสอ้างอิงแล้วให้ worker ไปดึงเอง --- ## 10. Materialized view สำหรับ dashboard เรียลไทม์ ### สถานการณ์ ต้องการหน้าจอที่แสดงยอดสรุปแบบเรียลไทม์ เช่นจำนวนรายการต่อสถานะ ยอดรวมรายชั่วโมง หรืออัตราความสำเร็จของแต่ละปลายทาง โดยไม่ยิง query หนัก ๆ ใส่ฐานข้อมูลหลักทุกครั้งที่มีคนเปิดหน้าจอ ### เดิมต้องใช้กี่ระบบ ```mermaid flowchart LR subgraph OLD["แบบเดิม ต้องมีสี่ระบบ"] K["Kafka"] --> F["Flink หรือ Kafka Streams"] F --> R["Redis เก็บผลสรุป"] R --> WS["WebSocket server แยกอีกตัว"] WS --> B1["Browser"] end subgraph NEW["แบบ Kore ระบบเดียว"] S["STREAM ของ event"] --> IC["consumer ภายในที่ทำหน้าที่สรุป"] IC --> T["TABLE ที่เก็บผลสรุป"] T -->|"WATCH"| GW2["gateway ที่แปลงเป็น WebSocket"] GW2 --> B2["Browser"] end ``` ### การไหลของข้อมูล ```mermaid sequenceDiagram autonumber participant SR as ระบบต้นทาง participant ST as STREAM ชื่อ txn.events participant AG as Aggregator consumer participant TB as TABLE ชื่อ dash.summary participant WS as WebSocket gateway participant BR as Browser BR->>WS: เปิดหน้าจอ WS->>TB: อ่านค่าสรุปปัจจุบันทั้งหมด TB-->>WS: ค่าปัจจุบัน WS-->>BR: วาดหน้าจอครั้งแรก WS->>TB: WATCH ตั้งแต่ offset ปัจจุบัน SR->>ST: append event ใหม่ ST->>AG: ส่งให้ aggregator AG->>TB: อัปเดตตัวนับที่เกี่ยวข้อง TB->>WS: push การเปลี่ยนแปลง WS-->>BR: ส่งเฉพาะส่วนที่เปลี่ยน ไม่ต้องส่งใหม่ทั้งหน้า ``` ### ข้อดีที่มาจากการที่ทุกอย่างอยู่บน log เดียวกัน - **สร้าง view ใหม่ได้โดยไม่ต้องรอข้อมูลใหม่** ถ้าอยากได้มุมมองที่ไม่เคยเก็บไว้ ให้สร้าง consumer ใหม่แล้วอ่านตั้งแต่ offset 0 ระบบจะไล่คำนวณย้อนหลังให้เอง - **แก้สูตรผิดแล้วคำนวณใหม่ได้** ลบ TABLE ปลายทางทิ้ง แล้วให้ consumer อ่านซ้ำจากต้น ซึ่งเป็นสิ่งที่ทำไม่ได้ถ้าเก็บแค่ตัวเลขสรุป - Aggregator อัปเดต TABLE และ commit offset ของตัวเองใน batch เดียวกัน จึงไม่มีทางนับซ้ำหรือนับขาดเมื่อ restart ### ต้องระวัง - ถ้ามีคนเปิดหน้าจอพร้อมกันหลายพันคน ให้ gateway ตัวเดียว WATCH แล้วกระจายต่อ อย่าให้ทุก browser มี WATCH ของตัวเอง - ตัวนับที่เปลี่ยนถี่มากควรรวมผลก่อนเขียน เช่นสะสมใน memory 200 มิลลิวินาทีแล้วค่อยเขียนครั้งเดียว ไม่งั้นจะเขียนหลายหมื่นครั้งต่อวินาทีโดยไม่จำเป็น # กลุ่มที่ 3 — งานเฉพาะทาง ## 11. Telemetry ingestion และการย่อข้อมูล ### สถานการณ์ อุปกรณ์หรือ service จำนวนมากส่งค่าวัดเข้ามาถี่ ๆ ข้อมูลดิบมีค่าแค่ช่วงสั้น ๆ แต่ค่าสรุปต้องเก็บยาว การเก็บดิบทั้งหมดไว้เป็นปีคือการเผาเงินเปล่า ### ออกแบบเป็นชั้นตามอายุข้อมูล ```mermaid flowchart LR DEV["ผู้ส่งข้อมูลจำนวนมาก"] -->|"เขียนเป็น batch ไม่ใช่ทีละค่า"| RAW["STREAM ชื่อ metrics.raw เก็บ 24 ชั่วโมง"] RAW --> AG1["consumer ย่อเป็นค่าเฉลี่ยรายนาที"] AG1 --> M1["STREAM ชื่อ metrics.1m เก็บ 30 วัน"] M1 --> AG2["consumer ย่อเป็นค่าเฉลี่ยรายชั่วโมง"] AG2 --> M2["STREAM ชื่อ metrics.1h เก็บ 2 ปี"] RAW --> LAT["consumer อัปเดตค่าล่าสุดลง TABLE ชื่อ metrics.latest"] LAT --> DASH["dashboard อ่านค่าปัจจุบันได้ทันทีโดยไม่ต้องสแกน stream"] N8>"ค่าล่าสุดควรอยู่ใน TABLE ส่วนประวัติควรอยู่ใน STREAM การเก็บทั้งสองแบบจากแหล่งเดียวกันคือสิ่งที่ระบบแยกทำได้ยาก"] LAT --- N8 ``` ### ตารางการตั้งค่าตามชั้น | Namespace | Retention | ขนาดต่อวันโดยประมาณ | Durability | |---|---|---|---| | `metrics.raw` | 24 ชั่วโมง | ใหญ่ที่สุด | `async` เพราะเสียไปบ้างไม่กระทบ | | `metrics.1m` | 30 วัน | ประมาณ 1 ใน 60 ของดิบ | `async` | | `metrics.1h` | 2 ปี | ประมาณ 1 ใน 3600 ของดิบ | `quorum` เพราะใช้รายงาน | | `metrics.latest` | ไม่มี เป็น TABLE | เท่ากับจำนวน series | `none` เพราะสร้างใหม่ได้จาก raw | ### เทคนิคที่ช่วยมาก - **ให้ผู้ส่งรวมค่าก่อนส่ง** ส่งทีละ 100 ค่าดีกว่าส่งทีละค่า 100 ครั้ง ลด overhead ต่อ record ได้หลายสิบเท่า - ใช้ **partition key เป็นรหัสอุปกรณ์** เพื่อให้ค่าของอุปกรณ์เดียวกันเรียงกันและบีบอัดได้ดี เพราะค่าที่ใกล้กันมักคล้ายกัน - เปิด compression ที่ระดับ batch จะลดขนาดได้มากในข้อมูลชนิดนี้ ### ต้องระวัง - ข้อมูลที่มาช้ากว่ากำหนด เช่นอุปกรณ์ที่ออฟไลน์แล้วส่งย้อนหลัง จะทำให้ค่าสรุปที่คำนวณไปแล้วไม่ตรง ต้องตัดสินใจตั้งแต่ออกแบบว่าจะยอมทิ้ง หรือจะคำนวณช่วงนั้นใหม่ - อย่าตั้งจำนวน partition น้อยเกินไปสำหรับ workload แบบนี้ เพราะขยายทีหลังจะทำให้การกระจาย key เปลี่ยน --- ## 12. โหมด embedded สำหรับแอปเดียวจบ ### สถานการณ์ มีแอปที่ต้องการคิวงานและ cache แต่การรัน Kafka หรือ Redis เพิ่มไม่คุ้มเลย เช่น - เครื่องมือภายในที่รันบน VPS เครื่องเดียว - ระบบที่ต้องไปติดตั้งที่หน้างานลูกค้า ซึ่งยิ่งมีชิ้นส่วนน้อยยิ่งดี - งานที่รันแบบ edge หรืออยู่หลัง firewall ที่เข้าถึงยาก - เครื่องมือที่จะแจกให้คนอื่นใช้ แล้วไม่อยากให้เขาต้องติดตั้งอะไรก่อน ### เทียบสองแบบ ```mermaid flowchart TB subgraph SEP["แบบต้องรันแยก"] AP1["แอปของเรา"] -->|"ผ่านเครือข่าย"| RD1["Redis"] AP1 -->|"ผ่านเครือข่าย"| KF1["Kafka หรือ RabbitMQ"] RD1 --> OPS1["ต้องมีคนดูแล ต้องมี backup ต้องมี monitoring"] KF1 --> OPS1 end subgraph EMB["แบบ embedded"] AP2["แอปของเรา ที่ import แพ็กเกจ embed"] --> IN["Kore ทำงานอยู่ในโปรเซสเดียวกัน"] IN --> DIR[("โฟลเดอร์ข้อมูลข้าง ๆ ตัวโปรแกรม")] AP2 --> OUT["แจกจ่ายเป็นไฟล์เดียว คัดลอกแล้วรันได้เลย"] end N9>"โหมดนี้คือสิ่งที่ Kafka ทำไม่ได้เลย และเป็นเหตุผลที่หนักแน่นที่สุดข้อหนึ่งของการเขียนระบบนี้เอง"] EMB --- N9 ``` ### โค้ดตัวอย่าง ```go // Everything below runs inside the application process. // No extra service to install, no network hop, no connection pool to tune. node, err := embed.Open(embed.Options{ DataDir: "./data", Listen: "", // empty means in-process only MaxMemory: 256 << 20, }) if err != nil { log.Fatal(err) } defer node.Close() // Cache with TTL, the same API the network client exposes. _ = node.Table("cache").Set(ctx, "rate:USDTHB", rate, 5*time.Minute) // Durable work queue that survives a restart. _ = node.Stream("jobs").Append(ctx, jobID, job) go node.Stream("jobs").Group("worker").Consume(ctx, func(m kore.Message) error { return process(m) // returning an error keeps the offset uncommitted for a retry }) ``` ### เส้นทางที่ดีของโหมดนี้ เริ่มแบบ embedded ตอนระบบยังเล็ก แล้วเมื่อโตขึ้นให้เปิด `Listen` เป็น address จริง จากนั้นย้ายไปรันแยกเป็นโปรเซส โดย **โค้ดของแอปไม่ต้องแก้เลย** เพราะเรียก API ตัวเดียวกัน เปลี่ยนแค่วิธีสร้าง client จาก `embed.Open` เป็น `kore.Dial` ```mermaid flowchart LR A["เริ่มต้น รันในโปรเซสเดียว"] -->|"โตขึ้น เปิด port ให้แอปอื่นต่อได้"| B["ยังอยู่ในโปรเซสเดิม แต่รับ client ภายนอกด้วย"] B -->|"โตอีก แยกเป็นโปรเซสของตัวเอง"| C["รันเป็น service แยกบนเครื่องเดียวกัน"] C -->|"ต้องการความทนทาน"| D["ขยายเป็นคลัสเตอร์สามโหนด"] N10>"เส้นทางนี้ไม่ต้องเขียนโค้ดใหม่ในแต่ละขั้น ซึ่งต่างจากการเริ่มด้วย map ใน memory แล้ววันหนึ่งต้องเปลี่ยนไปใช้ Redis ทั้งระบบ"] A --- N10 ``` ### ต้องระวัง - โหมด embedded ไม่มีความทนทานระดับเครื่อง ถ้าเครื่องพัง ข้อมูลหายไปกับเครื่อง ต้องมี backup เป็นระยะ - แอปกับ Kore แย่งหน่วยความจำกันในโปรเซสเดียว ต้องตั้ง `MaxMemory` ให้ชัดและตั้ง `GOMEMLIMIT` ของทั้งโปรเซส - ถ้าแอปมี memory leak หรือ panic จะพา Kore ล่มไปด้วย ควรใช้กับแอปที่นิ่งพอสมควรแล้ว --- # สรุป ## 13. ตารางสรุปว่าแต่ละยูสเคสใช้เครื่องมืออะไร | ยูสเคส | STREAM | TABLE | EPHEMERAL | GROUP | WATCH | BATCH | TTL | |---|---|---|---|---|---|---|---| | 1. Idempotency และ rate limit | | ใช้ | | | | | ใช้ | | 2. Event backbone | ใช้ | | | ใช้ | | | | | 3. เขียน state กับ event พร้อมกัน | ใช้ | ใช้ | | ใช้ | | ใช้ | | | 4. สะพานเชื่อมระบบ legacy | ใช้ | ใช้ | | | | ใช้ | | | 5. Audit trail | ใช้ | ใช้ | | | | ใช้ | | | 6. สถานะร่วมของ gateway | | ใช้ | | | ใช้ | | ใช้ | | 7. Notification fan-out | ใช้ | ใช้ | | ใช้ | ใช้ | | ใช้ | | 8. Session และ token revocation | ใช้ | ใช้ | | | ใช้ | | ใช้ | | 9. Job queue และงานตั้งเวลา | ใช้ | ใช้ | | ใช้ | ใช้ | ใช้ | ใช้ | | 10. Materialized view | ใช้ | ใช้ | | ใช้ | ใช้ | ใช้ | | | 11. Telemetry | ใช้ | ใช้ | | ใช้ | | | | | 12. Embedded mode | ใช้ | ใช้ | | ใช้ | | | ใช้ | **ข้อสังเกตจากตาราง** ยูสเคสที่ให้คุณค่าสูงที่สุดคือกลุ่มที่ใช้ทั้ง STREAM และ TABLE พร้อมกันในธุรกรรมเดียว ซึ่งเป็นสิ่งที่ทำไม่ได้ถ้าแยกเป็น Kafka กับ Redis คนละระบบ นี่คือคำตอบที่จับต้องได้ของคำถามว่า "รวมกันแล้วได้อะไรมากกว่าแค่ประหยัดงาน ops" ### จัดกลุ่มตามความคุ้มค่าและความยาก ```mermaid flowchart TB subgraph EASY["เริ่มได้ทันที ใช้แค่ Phase 1 ถึง 3"] E1["1. Idempotency และ rate limit"] E2["8. Session และ token revocation"] E3["12. Embedded mode"] E4["11. Telemetry"] end subgraph MID["ต้องมี consumer group ครบ คือ Phase 2 ขึ้นไป"] M1["2. Event backbone"] M2["7. Notification fan-out"] M3["9. Job queue"] M4["10. Materialized view"] end subgraph HARD["ต้องมี atomic batch และ replication คือ Phase 4 ขึ้นไป"] H1["3. เขียน state กับ event พร้อมกัน"] H2["4. สะพานเชื่อมระบบ legacy"] H3["5. Audit trail"] H4["6. สถานะร่วมของ gateway"] end EASY --> MID --> HARD N11>"ยูสเคสในกล่องแรกให้คุณค่าได้ตั้งแต่ยังไม่มี replication จึงเหมาะเป็นสนามทดลองจริงก่อนลงทุนกับส่วนที่ยาก"] EASY --- N11 ``` --- ## 14. สิ่งที่ไม่ควรเอา Kore ไปทำ สำคัญพอ ๆ กับการรู้ว่าเอาไปทำอะไรได้ คือรู้ว่าไม่ควรเอาไปทำอะไร | งาน | ทำไมไม่เหมาะ | ควรใช้อะไรแทน | |---|---|---| | ฐานข้อมูลหลักที่มีความสัมพันธ์ซับซ้อน | ไม่มี join ไม่มี foreign key ไม่มี query planner | PostgreSQL หรือฐานข้อมูลเชิงสัมพันธ์ | | ค้นหาข้อความแบบเต็ม | ไม่มี inverted index ไม่มีการจัดอันดับผลลัพธ์ | OpenSearch หรือ Meilisearch | | รายงานเชิงวิเคราะห์ที่สแกนข้อมูลย้อนหลังหลายปี | log ออกแบบมาให้อ่านตามลำดับ ไม่ได้ออกแบบมาให้ scan แบบคอลัมน์ | ClickHouse หรือ DuckDB หรือ data warehouse | | เก็บไฟล์ขนาดใหญ่ เช่นรูปหรือเอกสาร | record ที่ใหญ่มากทำให้ latency ของทุกคนแย่ลง | S3 หรือ MinIO แล้วเก็บแค่ URL ใน Kore | | ธุรกรรมข้ามหลาย entity ที่ต้อง ACID เต็มรูปแบบ | รองรับแค่ atomic ภายใน Raft group เดียว | ฐานข้อมูลที่ทำ distributed transaction ได้ | | คำถามเชิงกราฟ เช่นหาความสัมพันธ์หลายชั้น | ไม่มีโครงสร้างรองรับ | Neo4j หรือทำใน SQL แบบ recursive | | เก็บถาวรระยะยาวมากในราคาถูกที่สุด | ดิสก์ของ node แพงกว่า object storage หลายเท่า | ย้าย segment เก่าไป S3 แล้วเก็บแค่ index ไว้ | **หลักการง่าย ๆ** Kore เก่งเรื่อง "ลำดับของเหตุการณ์" กับ "ค่าล่าสุดของ key" ถ้าคำถามของคุณไม่ใช่สองแบบนี้ แปลว่ากำลังใช้เครื่องมือผิดชิ้น --- ## 15. เส้นทางย้ายจากระบบเดิม อย่าย้ายทีเดียวทั้งหมด ให้ย้ายเป็นชั้นโดยเริ่มจากชั้นที่พังแล้วไม่เจ็บ ```mermaid flowchart TB S1["ขั้นที่ 1 ใช้แทน Redis เฉพาะงานที่หายได้ เช่น cache และ rate limit"] --> S2["ขั้นที่ 2 เขียนข้อมูลเข้าทั้งระบบเดิมและ Kore พร้อมกัน แล้วเทียบผลทุกวัน"] S2 --> S3["ขั้นที่ 3 ย้ายผู้อ่านทีละราย โดยยังเขียนสองที่อยู่"] S3 --> S4["ขั้นที่ 4 เมื่อผู้อ่านย้ายครบและตัวเลขตรงกันติดต่อกันหลายสัปดาห์ ค่อยหยุดเขียนระบบเดิม"] S4 --> S5["ขั้นที่ 5 เก็บระบบเดิมไว้อีกอย่างน้อยหนึ่งเดือนก่อนปิด เผื่อต้องย้อนกลับ"] N12>"ขั้นที่ 2 คือขั้นที่คนมักข้าม แต่เป็นขั้นที่จับความต่างเล็ก ๆ ได้ก่อนที่มันจะกลายเป็นปัญหาใหญ่ตอนตัดระบบจริง"] S2 --- N12 ``` ### ตัวชี้วัดที่ต้องเทียบระหว่างขั้นที่ 2 | ตัวชี้วัด | เกณฑ์ที่ควรผ่านก่อนไปขั้นถัดไป | |---|---| | จำนวน record ที่เขียนสำเร็จ | ต่างกันไม่เกิน 0 รายการ ต่อวัน | | ลำดับของ record ต่อ key | ตรงกัน 100 เปอร์เซ็นต์ | | latency ที่ percentile 99 | ไม่แย่กว่าระบบเดิม | | จำนวนครั้งที่ต้อง restart | เป็นศูนย์ ติดต่อกันอย่างน้อย 30 วัน | | ข้อมูลหายหลังจำลองไฟดับ | เป็นศูนย์ จากการทดสอบอย่างน้อย 100 ครั้ง | --- ## 16. ควรเริ่มจากยูสเคสไหนก่อน ถ้าจะเลือกทำจริงหนึ่งอย่างเพื่อพิสูจน์ว่าใช้ได้ ผมแนะนำตามลำดับนี้ 1. **ยูสเคสที่ 12 โหมด embedded** — เพราะทดสอบได้ในเครื่องตัวเอง ไม่กระทบใคร และให้ feedback เร็วที่สุดว่า API ที่ออกแบบไว้ใช้งานจริงแล้วรู้สึกอย่างไร 2. **ยูสเคสที่ 11 telemetry** — เพราะเป็นงานที่ข้อมูลหายไปบ้างไม่เป็นไร แต่ปริมาณสูงพอที่จะเจอปัญหา performance จริง 3. **ยูสเคสที่ 1 idempotency และ rate limit** — เริ่มจากส่วน rate limit ก่อนซึ่งหายได้ แล้วค่อยขยับไป idempotency เมื่อมั่นใจ 4. **ยูสเคสที่ 2 event backbone** — เมื่อ consumer group ทำงานนิ่งแล้ว ใช้กับสายข้อมูลที่ไม่วิกฤตก่อน 5. **ยูสเคสที่ 3 เขียน state กับ event พร้อมกัน** — เก็บไว้ท้ายสุด เพราะให้คุณค่าสูงที่สุดแต่ก็ต้องการความถูกต้องสูงที่สุดเช่นกัน **ข้อควรจำ** ยูสเคสที่ 3 ถึง 6 เป็นกลุ่มที่ทำให้การรวม Kafka กับ Redis เข้าด้วยกันคุ้มค่าจริง ๆ แต่ก็เป็นกลุ่มที่ต้องการความมั่นใจในความถูกต้องมากที่สุด อย่ารีบไปถึงตรงนั้นก่อนที่ยูสเคสง่าย ๆ จะรันต่อเนื่องได้หลายเดือนโดยไม่มีปัญหา